fix(admin): merge unreadable-config marker with isolated diagnostics

rustfs/rustfs#7225 and rustfs/backlog#2309 refactored the same ten
per-config arms of ExportBucketMetadata concurrently. Keep #7225's
`exported_bucket_config` helper, `export_internal_error`, the
`?diagnostic=true` mode with its reserved `_diagnostic/` namespace,
manifest and import preflight, and re-apply #2309's semantics on top.

The helper now returns `ExportConfigError`, which separates a
configuration that is stored but undecodable from a failure of this
build's own serialization. An ordinary export records the former in
`<bucket>/rustfs-unreadable-configs.json` and continues, so one bucket
cannot abort a whole-cluster export, and fails closed on the latter.
Diagnostic mode is unchanged: it reports both through the archive-wide
manifest with a fixed code and no parser detail.

The #7225 assertion that a complete export fails closed on unreadable
targets is replaced by assertions of the new behaviour; every other
assertion on both sides is kept.

The s3_error! baseline is remeasured at 1589 for the merged tree.

crates/ecstore/src/api/mod.rs

@@ -32,7 +32,7 @@ pub mod bucket {
     pub mod bucket_target_sys {
         pub use crate::bucket::bucket_target_sys::{
             AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError,
-            SsecPassthroughCapability, TargetClient, append_version_id_query,
+            SsecPassthroughCapability, TargetClient, UnreadableTargetsPolicy, append_version_id_query,
         };
     }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -369,6 +369,26 @@ struct SsecPassthroughRecord {
     recorded_at: Instant,
 }
 
+/// What a target write does when the bucket's persisted target set exists but
+/// cannot be decoded.
+///
+/// `docs/architecture/remote-credential-sealing-adr.md` forbids rewriting a
+/// configuration that could not be fully read, because re-serializing a
+/// partial in-memory view is the one mechanism by which a configured target
+/// really disappears. That rule guards against an *unintentional* overwrite,
+/// so an operator who names the hazard keeps a repair path
+/// (rustfs/backlog#2309); everything that does not name it stays refused.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum UnreadableTargetsPolicy {
+    /// Refuse the write with [`BucketTargetError::BucketRemoteTargetsUnreadable`].
+    #[default]
+    FailClosed,
+    /// Discard the unreadable set; the target being written becomes the whole
+    /// configuration. Reachable only from an admin request that asked for it
+    /// explicitly, and audited by the caller.
+    Replace,
+}
+
 #[derive(Debug, Default)]
 pub struct BucketTargetSys {
     pub arn_remotes_map: Arc<RwLock<HashMap<String, ArnTarget>>>,
@@ -791,20 +811,45 @@ impl BucketTargetSys {
         bucket: &str,
         target: &BucketTarget,
         update: bool,
+        unreadable_policy: UnreadableTargetsPolicy,
     ) -> Result<BucketTargets, BucketTargetError> {
         self.validate_target(bucket, target).await?;
 
-        let mut bucket_targets = match self.list_bucket_targets(bucket).await {
-            Ok(targets) => targets,
-            Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => BucketTargets::default(),
-            Err(err) => return Err(err),
-        };
+        let mut bucket_targets = self.targets_base_for_write(bucket, unreadable_policy).await?;
 
         Self::upsert_target_entry(&mut bucket_targets.targets, target, update)?;
 
         Ok(bucket_targets)
     }
 
+    /// The persisted target set a write merges into.
+    ///
+    /// An absent configuration starts from the empty set. An unreadable one is
+    /// refused, because re-serializing a partial view of a set this node could
+    /// not decode is how a configured target disappears for good — unless the
+    /// caller carries the operator's explicit
+    /// [`UnreadableTargetsPolicy::Replace`] opt-in, which discards it
+    /// deliberately (rustfs/backlog#2309).
+    async fn targets_base_for_write(
+        &self,
+        bucket: &str,
+        unreadable_policy: UnreadableTargetsPolicy,
+    ) -> Result<BucketTargets, BucketTargetError> {
+        match self.list_bucket_targets(bucket).await {
+            Ok(targets) => Ok(targets),
+            Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => Ok(BucketTargets::default()),
+            // The opt-in discards only a set this node genuinely cannot read.
+            // A readable set still merges through the arm above, so the policy
+            // can never drop a target that was visible here.
+            Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
+                if unreadable_policy == UnreadableTargetsPolicy::Replace =>
+            {
+                Ok(BucketTargets::default())
+            }
+            Err(err) => Err(err),
+        }
+    }
+
     pub async fn validate_target(&self, bucket: &str, target: &BucketTarget) -> Result<(), BucketTargetError> {
         if !target.target_type.is_valid() {
             return Err(BucketTargetError::BucketRemoteArnTypeInvalid {
@@ -4313,4 +4358,75 @@ mod tests {
         let window = LastMinuteLatency::new();
         assert_eq!(window.get_total().avg, Duration::from_secs(0));
     }
+
+    fn repair_target(bucket: &str, id: &str) -> BucketTarget {
+        BucketTarget {
+            source_bucket: bucket.to_string(),
+            endpoint: "remote.example.com".to_string(),
+            target_bucket: "remote".to_string(),
+            arn: format!("arn:rustfs:replication:us-east-1:{bucket}:{id}"),
+            target_type: BucketTargetType::ReplicationService,
+            region: "us-east-1".to_string(),
+            ..Default::default()
+        }
+    }
+
+    /// rustfs/backlog#2309: after rustfs/rustfs#7172 an undecodable
+    /// `bucket-targets.json` left the bucket with no API repair path at all.
+    /// The refusal is the default and stays the default; the operator's
+    /// explicit opt-in is the only thing that discards the set, and it starts
+    /// the replacement from empty rather than from a partial view of bytes
+    /// this node never decoded.
+    #[tokio::test]
+    async fn an_unreadable_target_set_is_replaced_only_with_the_explicit_opt_in() {
+        let sys = BucketTargetSys::default();
+        let bucket = "targets-repair-opt-in";
+        sys.mark_targets_unreadable(bucket).await;
+
+        assert!(
+            matches!(
+                sys.targets_base_for_write(bucket, UnreadableTargetsPolicy::FailClosed).await,
+                Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
+            ),
+            "without the opt-in an unreadable target set must still refuse the write"
+        );
+        assert_eq!(
+            UnreadableTargetsPolicy::default(),
+            UnreadableTargetsPolicy::FailClosed,
+            "a caller that says nothing must get the refusal"
+        );
+
+        let base = sys
+            .targets_base_for_write(bucket, UnreadableTargetsPolicy::Replace)
+            .await
+            .expect("the explicit opt-in must let an operator replace an unreadable set");
+        assert!(
+            base.is_empty(),
+            "the replacement must start from an empty set, never from a partial decode"
+        );
+    }
+
+    /// The opt-in is not a wipe switch. On a set this node can read, both
+    /// policies take the same merge path, so a stray `replace-unreadable=true`
+    /// cannot drop a visible target — which is what makes the flag safe to
+    /// repeat in an operator's repair script.
+    #[tokio::test]
+    async fn the_opt_in_never_discards_a_readable_target_set() {
+        let sys = BucketTargetSys::default();
+        let bucket = "targets-repair-readable";
+        let existing = repair_target(bucket, "keep");
+        sys.targets_map
+            .write()
+            .await
+            .insert(bucket.to_string(), vec![existing.clone()]);
+
+        for policy in [UnreadableTargetsPolicy::FailClosed, UnreadableTargetsPolicy::Replace] {
+            let base = sys
+                .targets_base_for_write(bucket, policy)
+                .await
+                .expect("a readable target set must be readable under either policy");
+            assert_eq!(base.targets.len(), 1, "{policy:?} must keep the persisted target");
+            assert_eq!(base.targets[0].arn, existing.arn, "{policy:?} must not rewrite the persisted target");
+        }
+    }
 }

crates/ecstore/src/bucket/metadata.rs

@@ -1611,6 +1611,34 @@ mod test {
         assert!(bm.bucket_target_config.is_none());
     }
 
+    /// rustfs/backlog#2309: the MinIO-origin `.metadata.bin` this repository
+    /// already carries as a compatibility fixture stores
+    /// `BucketTargetsConfigJSON` as a bare JSON array, which `BucketTargets`
+    /// (a `{"targets":[…]}` struct with no array fallback) cannot decode. The
+    /// bytes below are the exact payload the fixture in
+    /// `metadata_test.rs::TEST_BUCKET_METADATA_HEX` decodes to, so if RustFS
+    /// ever grows the array-shaped compatibility parse, this test is where the
+    /// upgrade break is pinned and where the decision has to be recorded.
+    #[test]
+    fn minio_array_shaped_bucket_targets_are_unreadable() {
+        let minio_array = br#"[{"endpoint":"http://target.example.com","targetBucket":"tb","region":"us-east-1"}]"#.to_vec();
+        let mut bm = BucketMetadata::new("minio-array-targets");
+        bm.bucket_targets_config_json = minio_array.clone();
+
+        bm.parse_all_configs()
+            .expect("a MinIO-shaped targets blob must not fail the whole metadata load");
+
+        assert!(
+            bm.bucket_targets_unreadable(),
+            "an array-shaped MinIO targets blob is unreadable, not an empty target set"
+        );
+        assert!(bm.bucket_target_config.is_none());
+        assert_eq!(
+            bm.bucket_targets_config_json, minio_array,
+            "the raw MinIO bytes must survive so the configuration stays recoverable"
+        );
+    }
+
     /// 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

docs/operations/bucket-metadata-recovery.md

@@ -1,12 +1,12 @@
 # Bucket metadata diagnostics and recovery
 
-`GET /rustfs/admin/v3/export-bucket-metadata` keeps its strict behavior: an unreadable configuration fails the export. The optional `bucket` query selects one bucket; omitting it selects all buckets.
+`GET /rustfs/admin/v3/export-bucket-metadata` exports every configuration it can read. A configuration that is stored but unreadable is never exported and never replaced by a fabricated default; instead the bucket gains an entry `<bucket>/rustfs-unreadable-configs.json` of the shape `{"bucket": …, "unreadable": [{"config": …, "error": …}]}` naming each configuration that could not be read and why, and the export continues, so one bucket's undecodable payload cannot cost an operator the whole-cluster backup (rustfs/backlog#2309). The marker name is outside the configuration namespace importers dispatch on, so importing the archive back leaves the affected bucket's stored bytes untouched. A failure of the server's own serialization or archive writing still fails the export. The optional `bucket` query selects one bucket; omitting it selects all buckets.
 
-To inspect readable configurations while identifying failures, use the same authenticated endpoint with `?diagnostic=true`. This requires the existing `ExportBucketMetadataAction` permission. A successful response has:
+To collect a shareable support artifact that identifies the failures without carrying parser detail, use the same authenticated endpoint with `?diagnostic=true`. This requires the existing `ExportBucketMetadataAction` permission. A successful response has:
 
 - Filename `bucket-meta-diagnostic.zip` and header `x-rustfs-bucket-metadata-export: diagnostic`.
 - Readable entries under `_diagnostic/<bucket>/<config>`; target credentials remain redacted.
-- `_diagnostic-manifest.json`, containing `version: 1`, `mode: "diagnostic"`, `complete`, and an `errors` array. Each error identifies `bucket`, `config`, and the fixed code `configuration_unavailable`. The archive excludes unreadable payloads and parser error details.
+- `_diagnostic-manifest.json`, containing `version: 1`, `mode: "diagnostic"`, `complete`, and an `errors` array. Each error identifies `bucket`, `config`, and the fixed code `configuration_unavailable`. The archive excludes unreadable payloads and parser error details, and therefore carries no `rustfs-unreadable-configs.json` marker: the manifest reports the same failures with less detail, which is what makes a diagnostic archive safe to hand out.
 
 `complete` reports whether all supported configuration reads succeeded. A diagnostic archive is never a restorable backup, including when `complete` is true. Import rejects the manifest or reserved directory before any bucket creation or configuration write. The reserved directory is not a valid bucket name, so older importers cannot restore diagnostic entries as ordinary bucket configurations.
 
@@ -14,9 +14,11 @@ To inspect readable configurations while identifying failures, use the same auth
 
 RustFS currently accepts the documented `{"targets": [...]}` object format. It cannot decrypt MinIO KMS-encrypted target metadata. Unreadable target payloads remain failures instead of being interpreted as an empty target set; diagnostic export and replacement import do not add MinIO KMS decryption support.
 
-1. Inspect the diagnostic manifest to identify affected buckets. Preserve a separate backup of the original source configuration and any credentials needed for recovery.
+1. Inspect the diagnostic manifest, or the `rustfs-unreadable-configs.json` marker in an ordinary export, to identify affected buckets. Preserve a separate backup of the original source configuration and any credentials needed for recovery.
 2. Prepare a ZIP containing `<bucket>/bucket-targets.json` with a valid RustFS replacement, whose top-level shape is `{"targets": [...]}`. Supply the intended target settings and credentials; exported credentials are redacted. Use `{"targets": []}` only when intentionally clearing all targets, and reconcile any replication rules that reference removed targets.
 3. Submit the ZIP to the existing authenticated `PUT /rustfs/admin/v3/import-bucket-metadata` endpoint with `ImportBucketMetadataAction` permission. Import validates the replacement and persists it against the bucket incarnation; it does not need to parse the old target payload successfully.
-4. Verify target listing and the intended replication configuration. Retry the ordinary strict metadata export to confirm the unreadable configuration no longer blocks it.
+4. Verify target listing and the intended replication configuration. Retry the ordinary metadata export to confirm the bucket no longer carries an unreadable marker.
+
+Alternatively, `PUT /rustfs/admin/v3/set-remote-target?replace-unreadable=true` discards an undecodable target set as part of setting a replacement target. The flag is the operator's explicit acknowledgement that the stored set is being thrown away; without it the request is refused rather than rewriting an unreadable set from a partial view.
 
 Do not submit the diagnostic archive itself to the import endpoint. Copy only reviewed replacement entries into an ordinary import archive.

rustfs/src/admin/handlers/bucket_meta.rs

@@ -67,6 +67,17 @@ use zip::{ZipArchive, ZipWriter, write::SimpleFileOptions};
 const DIAGNOSTIC_EXPORT_PREFIX: &str = "_diagnostic";
 const DIAGNOSTIC_EXPORT_MANIFEST: &str = "_diagnostic-manifest.json";
 
+/// Archive entry naming the configurations a bucket stores but this build
+/// could not read (rustfs/backlog#2309).
+///
+/// The name deliberately sits outside the configuration-file namespace the
+/// importers switch on — `ImportBucketMetadata` matches known configuration
+/// names and ignores everything else — so no importer can mistake the marker
+/// for a configuration. Ordinary exports carry it; diagnostic exports report
+/// the same failures through [`DIAGNOSTIC_EXPORT_MANIFEST`] instead, which
+/// deliberately withholds the parser detail this marker records.
+const EXPORT_UNREADABLE_MANIFEST: &str = "rustfs-unreadable-configs.json";
+
 const LOG_COMPONENT_ADMIN: &str = "admin";
 const LOG_SUBSYSTEM_BUCKET_META: &str = "bucket_meta";
 const EVENT_ADMIN_BUCKET_META_STATE: &str = "admin_bucket_meta_state";
@@ -76,31 +87,82 @@ fn export_internal_error(message: impl Into<String>) -> s3s::S3Error {
     s3_error!(InternalError, "{message}")
 }
 
-fn checked_raw_xml<T, E, F>(validated: &T, raw: Vec<u8>, parse: F) -> S3Result<Vec<u8>>
+/// One configuration that is stored for a bucket but could not be exported.
+#[derive(serde::Serialize)]
+struct UnreadableExportEntry {
+    config: &'static str,
+    error: String,
+}
+
+#[derive(serde::Serialize)]
+struct UnreadableExportManifest<'a> {
+    bucket: &'a str,
+    unreadable: &'a [UnreadableExportEntry],
+}
+
+/// Why one of a bucket's configurations could not be exported.
+///
+/// The two variants are what an ordinary export dispatches on: a bucket whose
+/// stored bytes this build cannot turn into a configuration is named and
+/// skipped, while a failure of our own output machinery still fails the whole
+/// export closed.
+#[derive(Debug)]
+enum ExportConfigError {
+    /// The configuration is stored but this build cannot read it: a
+    /// MinIO-origin or otherwise undecodable blob, or a revision that moved
+    /// underneath the export. No retry of ours turns those bytes into a
+    /// configuration, so one such bucket must not abort a whole-cluster export
+    /// (rustfs/backlog#2309).
+    Unreadable(String),
+    /// This build failed to produce its own output for a configuration it had
+    /// already decoded. Nothing about the stored bytes is in doubt, so the
+    /// export fails closed rather than reporting healthy metadata as
+    /// unreadable.
+    Internal(s3s::S3Error),
+}
+
+impl ExportConfigError {
+    fn unreadable(message: impl Into<String>) -> Self {
+        Self::Unreadable(message.into())
+    }
+
+    fn internal(message: impl Into<String>) -> Self {
+        Self::Internal(export_internal_error(message))
+    }
+}
+
+fn checked_raw_xml<T, E, F>(validated: &T, raw: Vec<u8>, parse: F) -> Result<Vec<u8>, ExportConfigError>
 where
     T: PartialEq,
     E: std::fmt::Display,
     F: FnOnce(&[u8]) -> Result<T, E>,
 {
-    let selected = parse(&raw)
-        .map_err(|e| export_internal_error(format!("persisted bucket metadata changed to invalid XML during export: {e}")))?;
+    let selected = parse(&raw).map_err(|e| {
+        ExportConfigError::unreadable(format!("persisted bucket metadata changed to invalid XML during export: {e}"))
+    })?;
     if selected != *validated {
-        return Err(export_internal_error("bucket metadata changed during export"));
+        return Err(ExportConfigError::unreadable("bucket metadata changed during export"));
     }
     Ok(raw)
 }
 
-fn checked_versioning_xml(validated: &VersioningConfiguration, raw: Vec<u8>) -> S3Result<Vec<u8>> {
+fn checked_versioning_xml(validated: &VersioningConfiguration, raw: Vec<u8>) -> Result<Vec<u8>, ExportConfigError> {
     if raw.is_empty() {
         if *validated != VersioningConfiguration::default() {
-            return Err(export_internal_error("bucket metadata changed during export"));
+            return Err(ExportConfigError::unreadable("bucket metadata changed during export"));
         }
-        return serialize(validated).map_err(|e| export_internal_error(format!("serialize config failed: {e}")));
+        return serialize(validated).map_err(|e| ExportConfigError::internal(format!("serialize config failed: {e}")));
     }
     checked_raw_xml(validated, raw, deserialize::<VersioningConfiguration>)
 }
 
-async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec<u8>>> {
+/// Bytes to export for one of a bucket's configurations.
+///
+/// `Ok(None)` means the bucket has not configured it. An `Err` never becomes an
+/// exported configuration — a fabricated default here would reach an importer
+/// as a real one — and its variant tells the caller whether the failure belongs
+/// to the stored bytes or to this build's own output; see [`ExportConfigError`].
+async fn exported_bucket_config(bucket: &str, conf: &str) -> Result<Option<Vec<u8>>, ExportConfigError> {
     match conf {
         BUCKET_POLICY_CONFIG => {
             let config: BucketPolicy = match metadata_sys::get_bucket_policy(bucket).await {
@@ -109,11 +171,11 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "failed to load bucket metadata: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")));
                 }
             };
-            let config_json =
-                serde_json::to_vec(&config).map_err(|e| s3_error!(InternalError, "failed to serialize config: {e}"))?;
+            let config_json = serde_json::to_vec(&config)
+                .map_err(|e| ExportConfigError::internal(format!("failed to serialize config: {e}")))?;
             Ok(Some(config_json))
         }
         BUCKET_NOTIFICATION_CONFIG => {
@@ -123,14 +185,14 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
                 Ok(None) => return Ok(None),
             };
 
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("get bucket metadata failed: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .notification_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<s3s::dto::NotificationConfiguration>)?;
@@ -144,12 +206,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "failed to load bucket metadata: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("failed to load bucket metadata: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")))?
                 .lifecycle_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<BucketLifecycleConfiguration>)?;
@@ -163,12 +225,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "failed to load bucket metadata: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("failed to load bucket metadata: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")))?
                 .tagging_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<Tagging>)?;
@@ -182,11 +244,11 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
             let config_json =
-                serde_json::to_vec(&config).map_err(|e| s3_error!(InternalError, "serialize config failed: {e}"))?;
+                serde_json::to_vec(&config).map_err(|e| ExportConfigError::internal(format!("serialize config failed: {e}")))?;
 
             Ok(Some(config_json))
         }
@@ -197,12 +259,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("get bucket metadata failed: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .object_lock_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<ObjectLockConfiguration>)?;
@@ -216,12 +278,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("get bucket metadata failed: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .encryption_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<ServerSideEncryptionConfiguration>)?;
@@ -235,12 +297,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("get bucket metadata failed: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .versioning_config_xml
                 .clone();
             let config_xml = checked_versioning_xml(&config, raw_config)?;
@@ -254,12 +316,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
             let raw_config = metadata_sys::get(bucket)
                 .await
-                .map_err(|e| export_internal_error(format!("get bucket metadata failed: {e}")))?
+                .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .replication_config_xml
                 .clone();
             let config_xml = checked_raw_xml(&config, raw_config, deserialize::<ReplicationConfiguration>)?;
@@ -273,12 +335,12 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> S3Result<Option<Vec
                     if e == StorageError::ConfigNotFound {
                         return Ok(None);
                     }
-                    return Err(s3_error!(InternalError, "get bucket metadata failed: {e}"));
+                    return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
                 }
             };
 
             let config_json = serde_json::to_vec(&config.redacted_credentials())
-                .map_err(|e| s3_error!(InternalError, "serialize config failed: {e}"))?;
+                .map_err(|e| ExportConfigError::internal(format!("serialize config failed: {e}")))?;
 
             Ok(Some(config_json))
         }
@@ -377,19 +439,52 @@ impl Operation for ExportBucketMetadata {
         ];
 
         for bucket in buckets {
+            let mut unreadable: Vec<UnreadableExportEntry> = Vec::new();
             for &conf in confs.iter() {
                 let conf_path = path_join_buf(&[bucket.name.as_str(), conf]);
                 let config = match exported_bucket_config(&bucket.name, conf).await {
                     Ok(Some(config)) => config,
                     Ok(None) => continue,
-                    Err(error) if !query.diagnostic => return Err(error),
-                    Err(_) => {
-                        errors.push(serde_json::json!({
-                            "bucket": bucket.name,
-                            "config": conf,
-                            "code": "configuration_unavailable",
-                        }));
-                        continue;
+                    Err(error) => {
+                        if query.diagnostic {
+                            // A diagnostic archive names every configuration it
+                            // could not export under one fixed code, carrying
+                            // neither the payload nor the parser detail, so it
+                            // stays shareable (rustfs/rustfs#7225).
+                            errors.push(serde_json::json!({
+                                "bucket": bucket.name,
+                                "config": conf,
+                                "code": "configuration_unavailable",
+                            }));
+                            continue;
+                        }
+                        match error {
+                            // One bucket's undecodable blob must not abort the
+                            // whole-cluster export: record which configuration
+                            // could not be read and keep going, so an operator
+                            // migrating away still gets every readable
+                            // configuration (rustfs/backlog#2309).
+                            ExportConfigError::Unreadable(error) => {
+                                warn!(
+                                    event = EVENT_ADMIN_BUCKET_META_STATE,
+                                    component = LOG_COMPONENT_ADMIN,
+                                    subsystem = LOG_SUBSYSTEM_BUCKET_META,
+                                    action = "export_bucket_metadata",
+                                    result = "config_unreadable",
+                                    bucket = %bucket.name,
+                                    config_name = %conf,
+                                    error = %error,
+                                    "admin bucket meta state"
+                                );
+                                unreadable.push(UnreadableExportEntry { config: conf, error });
+                                continue;
+                            }
+                            // Our own encoder failed on a configuration this
+                            // build had already decoded. The stored bytes are
+                            // not in question, so fail the export instead of
+                            // reporting readable metadata as unreadable.
+                            ExportConfigError::Internal(error) => return Err(error),
+                        }
                     }
                 };
                 let conf_path = if query.diagnostic {
@@ -404,6 +499,23 @@ impl Operation for ExportBucketMetadata {
                     .write_all(&config)
                     .map_err(|e| s3_error!(InternalError, "failed to write archive entry: {e}"))?;
             }
+
+            // Only reachable outside diagnostic mode, which reports the same
+            // failures through the archive-wide manifest instead.
+            if !unreadable.is_empty() {
+                let manifest = serde_json::to_vec(&UnreadableExportManifest {
+                    bucket: bucket.name.as_str(),
+                    unreadable: &unreadable,
+                })
+                .map_err(|e| export_internal_error(format!("failed to serialize unreadable manifest: {e}")))?;
+                let manifest_path = path_join_buf(&[bucket.name.as_str(), EXPORT_UNREADABLE_MANIFEST]);
+                zip_writer
+                    .start_file(manifest_path, SimpleFileOptions::default())
+                    .map_err(|e| s3_error!(InternalError, "failed to start archive entry: {e}"))?;
+                zip_writer
+                    .write_all(&manifest)
+                    .map_err(|e| s3_error!(InternalError, "failed to write archive entry: {e}"))?;
+            }
         }
 
         if query.diagnostic {
@@ -1364,6 +1476,10 @@ mod backup_zip_compatibility_tests {
     const ROOT_ACCESS_KEY: &str = "BUCKETMETABACKUPROOT";
     const ROOT_SECRET_KEY: &str = "bucketMetaBackupRootSecret123";
     const BUCKET: &str = "backup-compatibility";
+    const UNREADABLE_BUCKET: &str = "minio-origin-targets";
+    /// The exact `BucketTargetsConfigJSON` payload carried by the MinIO
+    /// `.metadata.bin` fixture in `crates/ecstore/src/bucket/metadata_test.rs`.
+    const MINIO_ARRAY_TARGETS: &[u8] = br#"[{"endpoint":"http://target.example.com","targetBucket":"tb","region":"us-east-1"}]"#;
     const NOTIFICATION_XML: &[u8] = b"<NotificationConfiguration>\n</NotificationConfiguration>";
     const LIFECYCLE_XML: &[u8] = b"<LifecycleConfiguration>\n<Rule><ID>expire</ID><Status>Enabled</Status><Filter><Prefix>logs/</Prefix></Filter><Expiration><Days>30</Days></Expiration></Rule>\n</LifecycleConfiguration>";
     const SSE_XML: &[u8] = b"<ServerSideEncryptionConfiguration>\n<Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule>\n</ServerSideEncryptionConfiguration>";
@@ -1495,14 +1611,63 @@ mod backup_zip_compatibility_tests {
                 .expect("publish unreadable targets fixture");
             assert!(metadata_sys::get_bucket_targets_config(UNREADABLE).await.is_err());
 
-            let strict_error = ExportBucketMetadata {}
+            // rustfs/backlog#2309: an ordinary export no longer fails closed on
+            // a configuration that is stored but unreadable. It names that one
+            // configuration in the bucket's own marker entry and keeps every
+            // readable configuration of every bucket, so one MinIO-origin blob
+            // cannot cost an operator the whole-cluster backup.
+            let ordinary = ExportBucketMetadata {}
                 .call(
                     admin_request(Method::GET, Uri::from_static("/rustfs/admin/v3/export-bucket-metadata"), Vec::new()),
                     Params::new(),
                 )
                 .await
-                .expect_err("a complete export must fail closed on unreadable targets");
-            assert_eq!(*strict_error.code(), s3s::S3ErrorCode::InternalError);
+                .expect("one unreadable configuration must not abort the ordinary export");
+            assert_eq!(ordinary.output.0, StatusCode::OK);
+            assert!(!ordinary.headers.contains_key("x-rustfs-bucket-metadata-export"));
+            let ordinary_bytes = ordinary.output.1.collect().await.expect("read ordinary archive").to_bytes();
+            let mut ordinary_archive = ZipArchive::new(Cursor::new(&ordinary_bytes)).expect("open ordinary archive");
+            assert!(
+                ordinary_archive
+                    .by_name(&format!("{HEALTHY}/{BUCKET_VERSIONING_CONFIG}"))
+                    .is_ok(),
+                "a healthy bucket must still export while another bucket is unreadable"
+            );
+            assert!(
+                ordinary_archive
+                    .by_name(&format!("{HEALTHY}/{EXPORT_UNREADABLE_MANIFEST}"))
+                    .is_err(),
+                "a bucket whose configurations all read must carry no unreadable marker"
+            );
+            assert!(
+                ordinary_archive
+                    .by_name(&format!("{UNREADABLE}/{BUCKET_TARGETS_FILE}"))
+                    .is_err(),
+                "an unreadable targets blob must never be exported as a configuration"
+            );
+            let mut ordinary_marker = Vec::new();
+            ordinary_archive
+                .by_name(&format!("{UNREADABLE}/{EXPORT_UNREADABLE_MANIFEST}"))
+                .expect("the ordinary export must name the configuration it could not read")
+                .read_to_end(&mut ordinary_marker)
+                .expect("read unreadable marker");
+            assert!(
+                !ordinary_marker
+                    .windows(SECRET.len())
+                    .any(|window| window == SECRET.as_bytes())
+            );
+            let ordinary_marker: serde_json::Value = serde_json::from_slice(&ordinary_marker).expect("the marker must be JSON");
+            assert_eq!(ordinary_marker["bucket"], UNREADABLE);
+            assert_eq!(
+                ordinary_marker["unreadable"].as_array().map(Vec::len),
+                Some(1),
+                "only the configuration that could not be read may be marked: {ordinary_marker}"
+            );
+            assert_eq!(ordinary_marker["unreadable"][0]["config"], BUCKET_TARGETS_FILE);
+            assert!(
+                ordinary_marker["unreadable"][0]["error"].is_string(),
+                "the marker must carry the reason an operator needs to repair the bucket"
+            );
 
             let response = ExportBucketMetadata {}
                 .call(
@@ -1676,6 +1841,12 @@ mod backup_zip_compatibility_tests {
         assert!(archive.by_name(DIAGNOSTIC_EXPORT_MANIFEST).is_err());
         assert!(archive.by_name(&format!("{HEALTHY}/{BUCKET_VERSIONING_CONFIG}")).is_ok());
         assert!(archive.by_name(&format!("{UNREADABLE}/{BUCKET_TARGETS_FILE}")).is_ok());
+        assert!(
+            archive
+                .by_name(&format!("{UNREADABLE}/{EXPORT_UNREADABLE_MANIFEST}"))
+                .is_err(),
+            "the marker must disappear once the configuration reads again"
+        );
     }
 
     #[tokio::test]
@@ -1818,6 +1989,89 @@ mod backup_zip_compatibility_tests {
         }
         let _: ReplicationConfiguration =
             deserialize(&restored.replication_config_xml).expect("old parser must read the newly exported archive payload");
+
+        // rustfs/backlog#2309: a MinIO-origin `.metadata.bin` stores its
+        // targets as a bare JSON array, which `BucketTargets` cannot decode.
+        // Since rustfs/rustfs#7172 that reads as "stored but unreadable" —
+        // which must mark one bucket's one configuration, not abort the
+        // whole-cluster export an operator needs to migrate away.
+        env.make_bucket(UNREADABLE_BUCKET, false).await;
+        metadata_sys::update(UNREADABLE_BUCKET, BUCKET_TARGETS_FILE, MINIO_ARRAY_TARGETS.to_vec())
+            .await
+            .expect("persist the MinIO-shaped targets blob");
+        metadata_sys::get_bucket_targets_config(UNREADABLE_BUCKET)
+            .await
+            .expect_err("a MinIO array-shaped targets blob must read as unreadable, not as an empty set");
+
+        let cluster_export = ExportBucketMetadata {}
+            .call(
+                admin_request(Method::GET, Uri::from_static("/rustfs/admin/v3/export-bucket-metadata"), Vec::new()),
+                Params::new(),
+            )
+            .await
+            .expect("one bucket's unreadable configuration must not abort the whole-cluster export");
+        assert_eq!(cluster_export.output.0, StatusCode::OK);
+        let cluster_archive = cluster_export
+            .output
+            .1
+            .collect()
+            .await
+            .expect("read cluster archive body")
+            .to_bytes()
+            .to_vec();
+        let mut archive = ZipArchive::new(Cursor::new(&cluster_archive)).expect("open cluster archive");
+
+        // Every readable configuration of every other bucket still exports.
+        for (config_file, payload) in persisted_xml_fixtures() {
+            let mut exported_payload = Vec::new();
+            archive
+                .by_name(&format!("{BUCKET}/{config_file}"))
+                .unwrap_or_else(|_| panic!("cluster export must still contain {config_file}"))
+                .read_to_end(&mut exported_payload)
+                .unwrap_or_else(|_| panic!("read exported {config_file}"));
+            assert_eq!(exported_payload, payload, "one bad bucket must not change another bucket's export");
+        }
+        assert!(
+            archive.by_name(&format!("{BUCKET}/{EXPORT_UNREADABLE_MANIFEST}")).is_err(),
+            "a bucket whose configurations all read must carry no unreadable marker"
+        );
+
+        // The unreadable configuration is named rather than fabricated: no
+        // targets entry is exported for it at all.
+        assert!(
+            archive
+                .by_name(&format!("{UNREADABLE_BUCKET}/{BUCKET_TARGETS_FILE}"))
+                .is_err(),
+            "an unreadable targets blob must never be exported as a configuration"
+        );
+        let mut marker = Vec::new();
+        archive
+            .by_name(&format!("{UNREADABLE_BUCKET}/{EXPORT_UNREADABLE_MANIFEST}"))
+            .expect("the export must name the configuration it could not read")
+            .read_to_end(&mut marker)
+            .expect("read unreadable marker");
+        let marker: serde_json::Value = serde_json::from_slice(&marker).expect("the marker must be JSON");
+        assert_eq!(marker["bucket"], UNREADABLE_BUCKET);
+        assert_eq!(
+            marker["unreadable"].as_array().map(Vec::len),
+            Some(1),
+            "only the configuration that could not be read may be marked: {marker}"
+        );
+        assert_eq!(marker["unreadable"][0]["config"], BUCKET_TARGETS_FILE);
+        drop(archive);
+
+        // The marker cannot be misread as a configuration on the way back in:
+        // the importer switches on configuration names and ignores everything
+        // else, so the bucket's stored bytes come through untouched and the
+        // operator still has to repair them explicitly.
+        import_archive(cluster_archive).await;
+        let after_round_trip = metadata_sys::get_config_from_disk(UNREADABLE_BUCKET)
+            .await
+            .expect("the marked bucket must still load after the round trip");
+        assert_eq!(
+            after_round_trip.bucket_targets_config_json, MINIO_ARRAY_TARGETS,
+            "importing the marker must not overwrite or fabricate the bucket's targets configuration"
+        );
     }
 }
 

rustfs/src/admin/handlers/replication.rs

@@ -29,7 +29,7 @@ use crate::admin::storage_api::bucket::replication::{REMOTE_TARGET_READ_ONLY_HIS
 use crate::admin::storage_api::bucket::target::{
     BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos,
 };
-use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys};
+use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys, UnreadableTargetsPolicy};
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
 use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::storage_api::error::StorageError;
@@ -57,6 +57,20 @@ use url::Host;
 
 const SUPPORTED_REMOTE_TARGET_API: &str = "s3v4";
 
+const LOG_COMPONENT_ADMIN: &str = "admin";
+const LOG_SUBSYSTEM_REPLICATION: &str = "replication";
+const EVENT_ADMIN_REMOTE_TARGET_STATE: &str = "admin_remote_target_state";
+
+/// `set-remote-target?replace-unreadable=true`: the operator's explicit
+/// acknowledgement that this bucket's persisted target set cannot be decoded
+/// and is to be discarded (rustfs/backlog#2309).
+///
+/// Without it the refusal from rustfs/rustfs#7172 stands, which is what keeps
+/// an unreadable set from being silently rewritten from a partial view. The
+/// flag is deliberately absent from `PutBucketReplication`: targets are
+/// repaired first, then the rule is set.
+const REPLACE_UNREADABLE_TARGETS_PARAM: &str = "replace-unreadable";
+
 /// Field groups a `set-remote-target?update=true` request may modify, mirroring
 /// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update
 /// overlays only the requested groups onto the stored target, so a client can
@@ -541,6 +555,7 @@ impl Operation for SetRemoteTargetHandler {
         };
 
         let update = queries.get("update").is_some_and(|v| v == "true");
+        let replace_unreadable = queries.get(REPLACE_UNREADABLE_TARGETS_PARAM).is_some_and(|v| v == "true");
 
         warn!("set remote target, bucket: {}, update: {}", bucket, update);
 
@@ -708,10 +723,37 @@ impl Operation for SetRemoteTargetHandler {
 
         let arn = remote_target.arn.clone();
 
+        let unreadable_policy = if replace_unreadable {
+            UnreadableTargetsPolicy::Replace
+        } else {
+            UnreadableTargetsPolicy::FailClosed
+        };
+        let discarding_unreadable_targets = replace_unreadable
+            && matches!(
+                bucket_target_sys.list_bucket_targets(bucket).await,
+                Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
+            );
+
         let targets = bucket_target_sys
-            .set_target(bucket, &remote_target, update)
+            .set_target(bucket, &remote_target, update, unreadable_policy)
             .await
             .map_err(map_bucket_target_error)?;
+
+        // Audited only where the discard actually happened: the flag alone is
+        // not an event, on a readable set it changes nothing, and a refused
+        // write must not leave a record claiming the set was replaced.
+        if discarding_unreadable_targets {
+            warn!(
+                event = EVENT_ADMIN_REMOTE_TARGET_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                action = "set_remote_target",
+                result = "unreadable_targets_replaced",
+                bucket = %bucket,
+                arn = %remote_target.arn,
+                "admin remote target state"
+            );
+        }
         let json_targets = serde_json::to_vec(&targets).map_err(|e| {
             error!("Serialization error: {}", e);
             S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())

rustfs/src/admin/storage_api.rs

@@ -212,6 +212,7 @@ pub(crate) mod bucket_target_sys {
     pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError;
     pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability;
     pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient;
+    pub(crate) type UnreadableTargetsPolicy = super::ecstore_bucket::bucket_target_sys::UnreadableTargetsPolicy;
 }
 
 pub(crate) mod lifecycle {

rustfs/src/app/bucket_list_through.rs

@@ -38,7 +38,6 @@ use crate::on_demand_migration::{
     SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan,
 };
 use futures::StreamExt;
-use http::HeaderMap;
 use rustfs_utils::http::{SUFFIX_SOURCE_PROXY_REQUEST, get_header};
 use std::sync::Arc;
 use std::time::Instant;
@@ -476,6 +475,7 @@ mod tests {
         FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
         SourceCredentials, TlsConfig,
     };
+    use http::HeaderMap;
     use std::time::Duration;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 

scripts/check_s3s_footprint.sh

@@ -50,8 +50,13 @@ cd "$(dirname "$0")/.."
 # s3_error! stays flat at 1616.
 # 1616 → 1613 on 2026-09-02: dependency refresh verified the current tree has
 # already shed three s3_error! invocation lines; retighten the line counter.
+# 1613 → 1589 on 2026-09-06: rustfs/backlog#2309 and rustfs/rustfs#7225 both
+# folded the ten per-config arms of ExportBucketMetadata into one helper, which
+# now reports an unreadable configuration as a plain string instead of raising
+# an S3 error per arm (24 invocation lines removed from
+# rustfs/src/admin/handlers/bucket_meta.rs; measured after merging the two).
 S3S_IMPORT_FILES_BASELINE=213
-S3_ERROR_LINES_BASELINE=1613
+S3_ERROR_LINES_BASELINE=1589
 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
 # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
 # client was extracted to crates/s3-client, where s3s usage is legitimate;