fix(ecstore): fail closed on an unreadable bucket-targets blob (#7172)

* fix(ecstore): correct sealed-credential test helper parameter type

The helper took a HashMap that nothing imports, so the ecstore test target did not compile.

* fix(ecstore): fail closed on an unreadable bucket-targets blob

An undecodable bucket-targets.json was replaced by an empty BucketTargets,
so every replication target of that bucket disappeared, replication stopped,
and no caller saw an error. A missing secretKey alone triggers it, because
Credentials has no struct-level serde(default).

parse_all_configs now retains the failure instead: the raw bytes stay and the
typed field stays None, which BucketMetadata::bucket_targets_unreadable reads
as "exists but cannot be read" — the same distinction the fabricated marker
draws for bucket metadata as a whole. One corrupt sub-config still never fails
the metadata load, so an unreadable bucket cannot take down its neighbours or
the node.

BucketTargetSys records such buckets and answers every targets query with the
new BucketRemoteTargetsUnreadable, leaving any snapshot from an earlier
readable load in place so in-flight replication is not torn down. The
replication heal queue reports Missed rather than scheduling against an empty
target set, and the admin listing surfaces the fault instead of an empty list.

Refs: rustfs/backlog#2282

* fix(ecstore): report corrupt permissive bucket configs as invalid

Audit of the remaining parse_all_configs branches. Policy, versioning, object
lock and replication already fail closed at their accessors; encryption,
public access block and quota did not, and for those three "absent" is exactly
the state that grants something — plaintext storage, anonymous access,
unbounded capacity. They now report a stored-but-undecodable payload as
invalid rather than as ConfigNotFound, matching the guard the versioning and
object-lock accessors already use. The quota enforcement path already refused
such a payload; only the metadata read path was misreporting it.

The branches left degrading, and the concrete reason each is safe, are
recorded in the table on parse_all_configs.

Refs: rustfs/backlog#2282

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;