chore: integrate the current replication metadata boundary

Keep the batch on the frozen main baseline for final target validation.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -59,7 +59,7 @@ use rustfs_utils::http::{
     insert_header,
 };
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::error::Error;
 use std::fmt;
 use std::str::FromStr as _;
@@ -376,6 +376,11 @@ pub struct BucketTargetSys {
     /// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`.
     ssec_passthrough_map: Arc<RwLock<HashMap<String, SsecPassthroughRecord>>>,
     pub targets_map: Arc<RwLock<HashMap<String, Vec<BucketTarget>>>>,
+    /// Buckets whose persisted `bucket-targets.json` exists but cannot be
+    /// decoded (rustfs/backlog#2282). Written under the bucket's update mutex
+    /// alongside `targets_map`, and read before it so an unreadable
+    /// configuration surfaces as a typed error instead of an empty target set.
+    unreadable_targets: Arc<RwLock<HashSet<String>>>,
     pub h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
     target_h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
     pub hc_client: Arc<HttpClient>,
@@ -419,6 +424,7 @@ impl BucketTargetSys {
             arn_remotes_map: Arc::new(RwLock::new(HashMap::new())),
             ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())),
             targets_map: Arc::new(RwLock::new(HashMap::new())),
+            unreadable_targets: Arc::new(RwLock::new(HashSet::new())),
             h_mutex: Arc::new(RwLock::new(HashMap::new())),
             target_h_mutex: Arc::new(RwLock::new(HashMap::new())),
             hc_client: Arc::new(build_health_check_client()),
@@ -628,30 +634,40 @@ impl BucketTargetSys {
         health_map.clone()
     }
 
-    pub async fn list_targets(&self, bucket: &str, arn_type: &str) -> Vec<BucketTarget> {
+    /// Targets of one bucket, or of every bucket when `bucket` is empty.
+    ///
+    /// A bucket that simply has no targets yields an empty list; a bucket
+    /// whose persisted configuration cannot be decoded is an error, so an
+    /// admin listing reports the fault instead of an empty list that reads as
+    /// "replication is not configured" (rustfs/backlog#2282).
+    pub async fn list_targets(&self, bucket: &str, arn_type: &str) -> Result<Vec<BucketTarget>, BucketTargetError> {
         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.arn) {
-                            target.total_downtime = health.offline_duration;
-                            target.online = health.online;
-                            target.last_online = health.last_online;
-                            target.latency = target::LatencyStat {
-                                curr: health.latency.curr,
-                                avg: health.latency.avg,
-                                max: health.latency.peak,
-                            };
-                            target.offline_count = health.offline_count;
+            match self.list_bucket_targets(bucket).await {
+                Ok(bucket_targets) => {
+                    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.arn) {
+                                target.total_downtime = health.offline_duration;
+                                target.online = health.online;
+                                target.last_online = health.last_online;
+                                target.latency = target::LatencyStat {
+                                    curr: health.latency.curr,
+                                    avg: health.latency.avg,
+                                    max: health.latency.peak,
+                                };
+                                target.offline_count = health.offline_count;
+                            }
+                            targets.push(target);
                         }
-                        targets.push(target);
                     }
                 }
+                Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => {}
+                Err(err) => return Err(err),
             }
-            return targets;
+            return Ok(targets);
         }
 
         let targets_map = self.targets_map.read().await;
@@ -674,10 +690,16 @@ impl BucketTargetSys {
             }
         }
 
-        targets
+        Ok(targets)
     }
 
     pub async fn list_bucket_targets(&self, bucket: &str) -> Result<BucketTargets, BucketTargetError> {
+        if self.unreadable_targets.read().await.contains(bucket) {
+            return Err(BucketTargetError::BucketRemoteTargetsUnreadable {
+                bucket: bucket.to_string(),
+            });
+        }
+
         let targets_map = self.targets_map.read().await;
         if let Some(targets) = targets_map.get(bucket) {
             Ok(BucketTargets {
@@ -690,13 +712,30 @@ impl BucketTargetSys {
         }
     }
 
+    /// Record that this bucket's persisted targets configuration exists but
+    /// cannot be decoded (rustfs/backlog#2282).
+    ///
+    /// Any snapshot published from an earlier readable load is deliberately
+    /// left in place: withdrawing it would produce exactly the silent "no
+    /// targets configured" state this marker exists to prevent. The marker is
+    /// cleared by the next successful publish, which is what makes a repaired
+    /// configuration take effect without a restart.
+    pub async fn mark_targets_unreadable(&self, bucket: &str) {
+        let update_mutex = self.target_update_mutex(bucket).await;
+        let _update_guard = update_mutex.lock().await;
+
+        self.unreadable_targets.write().await.insert(bucket.to_string());
+    }
+
     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,
-        // then ssec_passthrough_map (always last; also taken standalone by the
-        // capability accessors).
+        // Lock order: unreadable_targets, then targets_map, then
+        // arn_remotes_map, then target_h_mutex, then ssec_passthrough_map
+        // (always last; also taken standalone by the capability accessors).
+        self.unreadable_targets.write().await.remove(bucket);
+
         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;
@@ -1093,6 +1132,11 @@ impl BucketTargetSys {
     /// Keeping persisted-config reads under the same mutex prevents a stale
     /// reload from overwriting a concurrent credential rotation.
     async fn update_all_targets_locked(&self, bucket: &str, targets: Option<&BucketTargets>) {
+        // Reaching here means the persisted configuration decoded, so the
+        // unreadable marker (if any) is stale. Cleared before the maps below
+        // so `unreadable_targets` stays the outermost of this module's locks.
+        self.unreadable_targets.write().await.remove(bucket);
+
         let mut clients = Vec::new();
         if let Some(new_targets) = targets {
             for target in &new_targets.targets {
@@ -1100,9 +1144,9 @@ impl BucketTargetSys {
             }
         }
 
-        // Lock order: targets_map, then arn_remotes_map, then target_h_mutex,
-        // then ssec_passthrough_map (always last; also taken standalone by the
-        // capability accessors).
+        // Lock order: unreadable_targets (above), then targets_map, then
+        // arn_remotes_map, then target_h_mutex, then ssec_passthrough_map
+        // (always last; also taken standalone by the capability accessors).
         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;
@@ -1161,6 +1205,11 @@ impl BucketTargetSys {
     }
 
     pub async fn set(&self, bucket: &str, meta: &BucketMetadata) {
+        if meta.bucket_targets_unreadable() {
+            self.mark_targets_unreadable(bucket).await;
+            return;
+        }
+
         let Some(config) = &meta.bucket_target_config else {
             return;
         };
@@ -2276,6 +2325,13 @@ pub enum BucketTargetError {
     BucketRemoteTargetNotFound {
         bucket: String,
     },
+    /// The bucket's persisted targets configuration exists but cannot be
+    /// decoded. Distinct from `BucketRemoteTargetNotFound`, which means the
+    /// bucket genuinely has no targets: callers must not degrade this one to
+    /// an empty target set (rustfs/backlog#2282).
+    BucketRemoteTargetsUnreadable {
+        bucket: String,
+    },
     BucketRemoteArnTypeInvalid {
         bucket: String,
     },
@@ -2309,6 +2365,9 @@ impl fmt::Display for BucketTargetError {
             BucketTargetError::BucketRemoteTargetNotFound { bucket } => {
                 write!(f, "Remote target not found for bucket: {bucket}")
             }
+            BucketTargetError::BucketRemoteTargetsUnreadable { bucket } => {
+                write!(f, "Persisted replication target configuration is unreadable for bucket: {bucket}")
+            }
             BucketTargetError::BucketRemoteArnTypeInvalid { bucket } => {
                 write!(f, "Invalid ARN type for bucket: {bucket}")
             }
@@ -3256,7 +3315,7 @@ mod tests {
             }],
         );
 
-        let targets = sys.list_targets("", "").await;
+        let targets = sys.list_targets("", "").await.expect("listing every bucket's targets");
 
         assert_eq!(targets.len(), 1);
         assert!(!targets[0].online);

crates/ecstore/src/bucket/metadata.rs

@@ -477,6 +477,18 @@ impl BucketMetadata {
         !self.table_bucket_config_json.is_empty()
     }
 
+    /// `bucket-targets.json` is stored for this bucket but this build cannot
+    /// decode it.
+    ///
+    /// Keeps "no replication targets configured" and "the target
+    /// configuration cannot be read" apart, the same distinction the
+    /// `fabricated` marker draws for the bucket metadata as a whole. Only
+    /// meaningful after [`Self::parse_all_configs`] has run; readers must fail
+    /// closed on `true` instead of serving an empty target set.
+    pub fn bucket_targets_unreadable(&self) -> bool {
+        !self.bucket_targets_config_json.is_empty() && self.bucket_target_config.is_none()
+    }
+
     /// Parsed per-bucket durability override, if a valid one is stored.
     ///
     /// Absent/empty/unparsable payloads all mean "no override" (the bucket
@@ -964,7 +976,32 @@ impl BucketMetadata {
         Ok(())
     }
 
-    fn parse_all_configs(&mut self) -> Result<()> {
+    /// Decode every stored sub-configuration into its typed field.
+    ///
+    /// A decode failure never fails the whole load: this runs on every bucket
+    /// metadata read, including startup and peer reload, so one bucket's
+    /// corrupt sub-configuration must not make the bucket — or the node —
+    /// unloadable. Instead the failure is *retained*: the raw bytes stay
+    /// untouched and the typed field stays `None`, so `!raw.is_empty() &&
+    /// typed.is_none()` is the durable "exists but cannot be read" signal that
+    /// each accessor keys off. Which accessors must fail closed on it:
+    ///
+    /// | Config | Verdict |
+    /// |---|---|
+    /// | policy | Fails closed: `get_bucket_policy` re-parses the raw JSON and propagates the error; `get_bucket_policy_raw` returns the stored bytes. |
+    /// | object lock | Fails closed in `object_lock_config_state_from_authoritative_metadata`; a retention decision may never be taken on a guess. |
+    /// | versioning | Fails closed in `get_versioning_config`; guessing Unversioned would make delete markers and version ids diverge from what is on disk. |
+    /// | replication | Fails closed in `get_replication_config`. |
+    /// | bucket targets | Fails closed in `get_bucket_targets_config`, and `sync_bucket_target_sys` marks the bucket unreadable in `BucketTargetSys` instead of publishing an empty target set (rustfs/backlog#2282). |
+    /// | encryption | Fails closed in `get_sse_config`: degrading to "no default encryption" stores plaintext objects the operator required to be encrypted. |
+    /// | public access block | Fails closed in `get_public_access_block_config`: degrading grants the anonymous access the operator asked to block. |
+    /// | quota | Fails closed in `get_quota_config`; the enforcement path in `quota::checker` already re-parses the raw JSON and refuses on error. |
+    /// | lifecycle | Safe to degrade: no rules means no expiration and no transition, so nothing is deleted or moved on the strength of an unreadable rule set. The bucket keeps serving reads and writes. |
+    /// | notification | Safe to degrade: events are an outbound side channel; no consumer draws a durability or authorization conclusion from their absence. |
+    /// | tagging | Safe to degrade: bucket tags are cost-allocation labels here; object-level tag conditions come from object metadata, not this blob. |
+    /// | CORS | Safe to degrade: an absent CORS configuration rejects cross-origin browser requests, which is already the restrictive direction. |
+    /// | logging, website, accelerate, request payment, bucket ACL | Safe to degrade: each only shapes an optional response or an optional side channel, and none of them authorizes an action or decides whether data is retained. |
+    pub(super) fn parse_all_configs(&mut self) -> Result<()> {
         if let Err(e) = self.parse_policy_config() {
             tracing::warn!(
                 event = "bucket_metadata_parse_failed",
@@ -1088,20 +1125,26 @@ impl BucketMetadata {
                 "Failed to parse bucket metadata config"
             );
         }
+        // A stored targets blob that cannot be decoded must not collapse into
+        // the empty target set: that is indistinguishable from "no replication
+        // configured", so replication stops and no caller ever sees an error
+        // (rustfs/backlog#2282). Leaving the typed field `None` while the raw
+        // bytes stay non-empty is the retained parse failure every targets
+        // reader keys off; the bytes are preserved so the configuration is
+        // still recoverable.
+        self.bucket_target_config = None;
         if !self.bucket_targets_config_json.is_empty() {
-            if let Err(e) = serde_json::from_slice::<BucketTargets>(&self.bucket_targets_config_json)
-                .map(|t| self.bucket_target_config = Some(t))
-            {
-                tracing::warn!(
+            match serde_json::from_slice::<BucketTargets>(&self.bucket_targets_config_json) {
+                Ok(targets) => self.bucket_target_config = Some(targets),
+                Err(e) => tracing::error!(
                     event = "bucket_metadata_parse_failed",
                     component = "ecstore",
                     subsystem = "bucket_metadata",
                     bucket = %self.name,
                     config = "bucket_targets",
                     error = %e,
-                    "Failed to parse bucket metadata config"
-                );
-                self.bucket_target_config = Some(BucketTargets::default());
+                    "Bucket replication targets are unreadable; replication for this bucket fails closed"
+                ),
             }
         } else {
             self.bucket_target_config = Some(BucketTargets::default());
@@ -1535,6 +1578,117 @@ mod test {
         assert_eq!(bucket_targets.targets[0].target_bucket, "target-bucket");
     }
 
+    /// rustfs/backlog#2282: a stored targets blob this build cannot decode
+    /// must not become the empty target set, and must stay distinguishable
+    /// from a bucket that never configured a target.
+    #[test]
+    fn unreadable_bucket_targets_never_degrade_to_an_empty_target_set() {
+        let truncated = br#"{"targets":[{"endpoint":"s3.example.com","#.to_vec();
+        let mut corrupt = BucketMetadata::new("corrupt-targets");
+        corrupt.bucket_targets_config_json = truncated.clone();
+
+        corrupt
+            .parse_all_configs()
+            .expect("one unreadable sub-config must not fail the whole metadata load");
+
+        assert!(
+            corrupt.bucket_target_config.is_none(),
+            "an undecodable targets blob must not produce a target set at all"
+        );
+        assert!(corrupt.bucket_targets_unreadable());
+        assert_eq!(
+            corrupt.bucket_targets_config_json, truncated,
+            "the raw bytes must survive so the configuration stays recoverable"
+        );
+
+        // The genuinely-absent case is unchanged, and the two now diverge.
+        let mut absent = BucketMetadata::new("no-targets");
+        absent.parse_all_configs().expect("absent targets parse");
+        assert!(
+            absent.bucket_target_config.as_ref().is_some_and(BucketTargets::is_empty),
+            "a bucket that configured no target still reads as an empty target set"
+        );
+        assert!(!absent.bucket_targets_unreadable());
+    }
+
+    /// `Credentials` carries no struct-level `serde(default)`, so one target
+    /// missing `secretKey` is a hard parse error for the whole document. That
+    /// must surface as "unreadable", never as "no targets configured".
+    #[test]
+    fn bucket_targets_missing_secret_key_are_unreadable_not_empty() {
+        let mut bm = BucketMetadata::new("missing-secret-key");
+        bm.bucket_targets_config_json = br#"{"targets":[{"endpoint":"s3.example.com","targetbucket":"remote","arn":"arn:rustfs:replication:us-east-1:src:1","credentials":{"accessKey":"AKIAEXAMPLE"}}]}"#.to_vec();
+
+        bm.parse_all_configs()
+            .expect("a rejected targets document must not fail the whole metadata load");
+
+        assert!(
+            bm.bucket_targets_unreadable(),
+            "a targets document rejected for a missing secretKey is unreadable, not empty"
+        );
+        assert!(bm.bucket_target_config.is_none());
+    }
+
+    /// The invariant every branch of `parse_all_configs` shares: a stored but
+    /// undecodable payload keeps its raw bytes and leaves the typed field
+    /// `None`, so no branch fabricates a value. What a reader may then do with
+    /// that state is decided per config; see the table on `parse_all_configs`.
+    #[test]
+    fn every_config_branch_retains_its_parse_failure_instead_of_defaulting() {
+        let malformed_xml = b"<not-a-valid-document".to_vec();
+        let malformed_json = b"{not-json".to_vec();
+
+        let mut bm = BucketMetadata::new("all-configs-malformed");
+        bm.policy_config_json = malformed_json.clone();
+        bm.quota_config_json = malformed_json.clone();
+        bm.bucket_targets_config_json = malformed_json.clone();
+        bm.notification_config_xml = malformed_xml.clone();
+        bm.lifecycle_config_xml = malformed_xml.clone();
+        bm.object_lock_config_xml = malformed_xml.clone();
+        bm.versioning_config_xml = malformed_xml.clone();
+        bm.encryption_config_xml = malformed_xml.clone();
+        bm.tagging_config_xml = malformed_xml.clone();
+        bm.replication_config_xml = malformed_xml.clone();
+        bm.cors_config_xml = malformed_xml.clone();
+        bm.logging_config_xml = malformed_xml.clone();
+        bm.website_config_xml = malformed_xml.clone();
+        bm.accelerate_config_xml = malformed_xml.clone();
+        bm.request_payment_config_xml = malformed_xml.clone();
+        bm.public_access_block_config_xml = malformed_xml.clone();
+        // `bucket_acl_config_json` is only checked for UTF-8, so only invalid
+        // UTF-8 exercises its failure branch.
+        bm.bucket_acl_config_json = vec![0xff, 0xfe];
+
+        bm.parse_all_configs()
+            .expect("a bucket whose every config is corrupt must still load its metadata");
+
+        let cleared: [(&str, bool); 17] = [
+            ("policy", bm.policy_config.is_none()),
+            ("quota", bm.quota_config.is_none()),
+            ("bucket_targets", bm.bucket_target_config.is_none()),
+            ("notification", bm.notification_config.is_none()),
+            ("lifecycle", bm.lifecycle_config.is_none()),
+            ("object_lock", bm.object_lock_config.is_none()),
+            ("versioning", bm.versioning_config.is_none()),
+            ("encryption", bm.sse_config.is_none()),
+            ("tagging", bm.tagging_config.is_none()),
+            ("replication", bm.replication_config.is_none()),
+            ("cors", bm.cors_config.is_none()),
+            ("logging", bm.logging_config.is_none()),
+            ("website", bm.website_config.is_none()),
+            ("accelerate", bm.accelerate_config.is_none()),
+            ("request_payment", bm.request_payment_config.is_none()),
+            ("public_access_block", bm.public_access_block_config.is_none()),
+            ("bucket_acl", bm.bucket_acl_config.is_none()),
+        ];
+        for (config, is_cleared) in cleared {
+            assert!(is_cleared, "{config}: a corrupt payload must not be replaced by a default");
+        }
+
+        assert_eq!(bm.bucket_targets_config_json, malformed_json, "raw bytes are retained");
+        assert_eq!(bm.lifecycle_config_xml, malformed_xml, "raw bytes are retained");
+    }
+
     #[test]
     fn lifecycle_update_config_clears_parsed_config_on_delete() {
         let mut bm = BucketMetadata::new("test-bucket");

crates/ecstore/src/bucket/metadata_sys.rs

@@ -360,6 +360,16 @@ async fn refresh_buckets_metadata_once(sys: Arc<RwLock<BucketMetadataSys>>) {
 }
 
 async fn sync_bucket_target_sys(bucket: &str, bm: &BucketMetadata) {
+    if bm.bucket_targets_unreadable() {
+        // "The configuration cannot be read" is not "no targets configured".
+        // Publishing an empty snapshot here is what silently stopped
+        // replication (rustfs/backlog#2282): mark the bucket instead, so every
+        // targets reader gets a typed error, and leave any snapshot from an
+        // earlier readable load in place rather than withdrawing it.
+        BucketTargetSys::get().mark_targets_unreadable(bucket).await;
+        return;
+    }
+
     BucketTargetSys::get()
         .update_all_targets(bucket, bm.bucket_target_config.as_ref())
         .await;
@@ -2118,7 +2128,9 @@ impl BucketMetadataSys {
     pub async fn get_public_access_block_config(&self, bucket: &str) -> Result<(PublicAccessBlockConfiguration, OffsetDateTime)> {
         let (bm, _) = self.get_config(bucket).await?;
 
-        if let Some(config) = &bm.public_access_block_config {
+        if !bm.public_access_block_config_xml.is_empty() && bm.public_access_block_config.is_none() {
+            Err(Error::other("persisted bucket public access block configuration is invalid"))
+        } else if let Some(config) = &bm.public_access_block_config {
             Ok((config.clone(), bm.public_access_block_config_updated_at))
         } else {
             Err(Error::ConfigNotFound)
@@ -2429,7 +2441,9 @@ impl BucketMetadataSys {
     pub async fn get_sse_config(&self, bucket: &str) -> Result<(ServerSideEncryptionConfiguration, OffsetDateTime)> {
         let (bm, _) = self.get_config(bucket).await?;
 
-        if let Some(config) = &bm.sse_config {
+        if !bm.encryption_config_xml.is_empty() && bm.sse_config.is_none() {
+            Err(Error::other("persisted bucket encryption configuration is invalid"))
+        } else if let Some(config) = &bm.sse_config {
             Ok((config.clone(), bm.encryption_config_updated_at))
         } else {
             Err(Error::ConfigNotFound)
@@ -2500,7 +2514,9 @@ impl BucketMetadataSys {
     pub async fn get_quota_config(&self, bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> {
         let (bm, _) = self.get_config(bucket).await?;
 
-        if let Some(config) = &bm.quota_config {
+        if !bm.quota_config_json.is_empty() && bm.quota_config.is_none() {
+            Err(Error::other("persisted bucket quota configuration is invalid"))
+        } else if let Some(config) = &bm.quota_config {
             Ok((config.clone(), bm.quota_config_updated_at))
         } else {
             Err(Error::ConfigNotFound)
@@ -2522,7 +2538,9 @@ impl BucketMetadataSys {
     pub async fn get_bucket_targets_config(&self, bucket: &str) -> Result<BucketTargets> {
         let (bm, _) = self.get_config(bucket).await?;
 
-        if let Some(config) = &bm.bucket_target_config {
+        if bm.bucket_targets_unreadable() {
+            Err(Error::other("persisted bucket replication target configuration is invalid"))
+        } else if let Some(config) = &bm.bucket_target_config {
             Ok(config.clone())
         } else {
             Err(Error::ConfigNotFound)
@@ -2593,6 +2611,7 @@ pub(crate) mod test_support {
 mod tests {
     use super::test_support::isolated_store_over_temp_disks;
     use super::*;
+    use crate::bucket::bucket_target_sys::BucketTargetError;
     use crate::bucket::metadata::{
         BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG, BUCKET_NOTIFICATION_CONFIG,
         BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_REQUEST_PAYMENT_CONFIG,
@@ -2788,6 +2807,36 @@ mod tests {
         );
     }
 
+    /// The `parse_all_configs` audit (rustfs/backlog#2282): every accessor
+    /// whose configuration grants something — plaintext storage, anonymous
+    /// access, capacity, replication targets — reports a corrupt payload as
+    /// invalid rather than as absent, because "absent" is what grants it.
+    #[tokio::test]
+    async fn malformed_permissive_configs_are_not_reported_as_absent() {
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore);
+        let bucket = "malformed-permissive-config";
+        let mut metadata = BucketMetadata::new(bucket);
+        metadata.encryption_config_xml = b"<ServerSideEncryptionConfiguration".to_vec();
+        metadata.public_access_block_config_xml = b"<PublicAccessBlockConfiguration".to_vec();
+        metadata.quota_config_json = b"{not-json".to_vec();
+        metadata.bucket_targets_config_json = b"{not-json".to_vec();
+        metadata
+            .parse_all_configs()
+            .expect("a corrupt sub-config must not fail the load");
+        sys.set(bucket.to_string(), Arc::new(metadata)).await;
+
+        for (config, result) in [
+            ("encryption", sys.get_sse_config(bucket).await.err()),
+            ("public access block", sys.get_public_access_block_config(bucket).await.err()),
+            ("quota", sys.get_quota_config(bucket).await.err()),
+            ("bucket targets", sys.get_bucket_targets_config(bucket).await.err()),
+        ] {
+            let err = result.unwrap_or_else(|| panic!("malformed {config} metadata must not read as a value"));
+            assert_ne!(err, Error::ConfigNotFound, "malformed {config} metadata must not be reported as absent");
+        }
+    }
+
     #[tokio::test]
     async fn config_states_distinguish_authoritative_absence_from_fabricated_metadata() {
         use std::sync::atomic::Ordering;
@@ -4066,6 +4115,114 @@ mod tests {
         target_sys.delete(bucket).await;
     }
 
+    /// rustfs/backlog#2282: an unreadable `bucket-targets.json` reaches every
+    /// targets reader as a typed error; it neither withdraws a snapshot a
+    /// previous readable load published, nor collapses into the "no targets
+    /// configured" state that a bucket with an absent configuration reports.
+    #[tokio::test]
+    #[serial]
+    async fn unreadable_bucket_targets_fail_closed_and_stay_distinct_from_absent() {
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore);
+        let target_sys = BucketTargetSys::get();
+        let unreadable = "targets-unreadable";
+        let absent = "targets-absent";
+        target_sys.delete(unreadable).await;
+        target_sys.delete(absent).await;
+
+        // A readable load publishes this bucket's targets.
+        let mut readable = BucketMetadata::new(unreadable);
+        readable.bucket_target_config = Some(BucketTargets {
+            targets: vec![target(unreadable, "live")],
+        });
+        sync_bucket_target_sys(unreadable, &readable).await;
+        assert_eq!(
+            target_sys
+                .list_bucket_targets(unreadable)
+                .await
+                .expect("readable targets publish")
+                .targets
+                .len(),
+            1
+        );
+
+        // The same bucket reloaded with a blob that cannot be decoded.
+        let mut corrupt = BucketMetadata::new(unreadable);
+        corrupt.bucket_targets_config_json = br#"{"targets":[{"endpoint":"#.to_vec();
+        corrupt
+            .parse_all_configs()
+            .expect("an unreadable targets blob must not fail the metadata load");
+        sys.set(unreadable.to_string(), Arc::new(corrupt)).await;
+
+        assert!(
+            matches!(
+                target_sys.list_bucket_targets(unreadable).await,
+                Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
+            ),
+            "an unreadable configuration must not read as an empty or a missing target set"
+        );
+        assert!(
+            target_sys.list_targets(unreadable, "").await.is_err(),
+            "the admin listing must surface the fault instead of an empty list"
+        );
+        let err = sys
+            .get_bucket_targets_config(unreadable)
+            .await
+            .expect_err("an unreadable targets configuration must not read as a value");
+        assert_ne!(err, Error::ConfigNotFound, "unreadable must not be reported as absent");
+
+        // A bucket that never configured a target keeps its previous behavior.
+        let mut no_targets = BucketMetadata::new(absent);
+        no_targets.parse_all_configs().expect("absent targets parse");
+        sys.set(absent.to_string(), Arc::new(no_targets)).await;
+        assert!(
+            matches!(
+                target_sys.list_bucket_targets(absent).await,
+                Err(BucketTargetError::BucketRemoteTargetNotFound { .. })
+            ),
+            "an absent configuration must still report as a missing target set"
+        );
+        assert!(
+            target_sys
+                .list_targets(absent, "")
+                .await
+                .expect("an absent configuration lists no targets")
+                .is_empty()
+        );
+        assert!(
+            sys.get_bucket_targets_config(absent)
+                .await
+                .expect("an absent targets configuration still reads as an empty set")
+                .is_empty(),
+            "the absent path must keep returning an empty target set, exactly as before"
+        );
+
+        // One bucket's unreadable configuration does not reach another bucket.
+        assert!(!matches!(
+            target_sys.list_bucket_targets(absent).await,
+            Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
+        ));
+
+        // A repaired configuration takes effect on the next load, no restart.
+        let mut repaired = BucketMetadata::new(unreadable);
+        repaired.bucket_target_config = Some(BucketTargets {
+            targets: vec![target(unreadable, "repaired")],
+        });
+        sync_bucket_target_sys(unreadable, &repaired).await;
+        assert_eq!(
+            target_sys
+                .list_bucket_targets(unreadable)
+                .await
+                .expect("a repaired configuration clears the unreadable marker")
+                .targets
+                .len(),
+            1
+        );
+
+        target_sys.delete(unreadable).await;
+        target_sys.delete(absent).await;
+    }
+
     #[tokio::test]
     #[serial]
     async fn metadata_reload_clears_stale_bucket_targets_when_config_is_removed() {

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

@@ -46,7 +46,7 @@ use super::replication_storage_boundary::{
     HTTPPreconditions, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO,
     ReplicationStorage,
 };
-use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted};
+use super::replication_target_boundary::{BucketTargetError, ReplicationTargetStore, replication_object_is_ssec_encrypted};
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
 use futures_util::stream::{self, StreamExt};
@@ -3084,6 +3084,23 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
 
     let tgts = match ReplicationTargetStore::list_bucket_targets(bucket).await {
         Ok(targets) => Some(targets),
+        // A bucket whose persisted target configuration cannot be decoded has
+        // an unknown target set, not an empty one: scheduling against `None`
+        // here would drop every heal for it without a trace
+        // (rustfs/backlog#2282). Report it missed so the object is retried
+        // once the configuration is readable again.
+        Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. }) => {
+            warn!(
+                event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket,
+                reason = "target_config_unreadable",
+                "Bucket replication targets are unreadable; replication heal queue fails closed"
+            );
+
+            return ReplicationQueueAdmission::Missed;
+        }
         Err(err) => {
             debug!(
                 event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED,

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

@@ -15,7 +15,8 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys};
+pub(crate) use crate::bucket::bucket_target_sys::BucketTargetError;
+use crate::bucket::bucket_target_sys::BucketTargetSys;
 use aws_sdk_s3::operation::head_object::HeadObjectOutput;
 use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode};
 use http::HeaderMap;

rustfs/src/admin/handlers/replication.rs

@@ -121,6 +121,11 @@ fn map_bucket_target_error(err: BucketTargetError) -> S3Error {
         | BucketTargetError::BucketRemoteRemoveDisallowed { .. } => {
             S3Error::with_message(S3ErrorCode::InvalidRequest, err.to_string())
         }
+        // A stored target configuration this node cannot decode is a
+        // server-side data fault, not a bad request (rustfs/backlog#2282).
+        BucketTargetError::BucketRemoteTargetsUnreadable { .. } => {
+            S3Error::with_message(S3ErrorCode::InternalError, err.to_string())
+        }
         BucketTargetError::Io(io_err) => S3Error::with_message(S3ErrorCode::InternalError, io_err.to_string()),
     }
 }
@@ -753,7 +758,12 @@ impl Operation for ListRemoteTargetHandler {
                 .map_err(ApiError::from)?;
 
             let sys = BucketTargetSys::get();
-            let targets = sys.list_targets(bucket, "").await;
+            // An unreadable targets configuration must not be reported as an
+            // empty target list (rustfs/backlog#2282).
+            let targets = sys.list_targets(bucket, "").await.map_err(|e| {
+                error!("list remote targets failed: {}", e);
+                map_bucket_target_error(e)
+            })?;
 
             let targets: Vec<_> = targets
                 .iter()