feat(ilm): add immutable legacy recovery exports

crates/ecstore/src/api/mod.rs

@@ -76,6 +76,13 @@ pub mod bucket {
             };
         }
 
+        pub mod recovery_export {
+            pub use crate::bucket::lifecycle::recovery_export::{
+                IlmRecoveryExportCreated, IlmRecoveryExportObservation, create_recovery_export,
+                inspect_recovery_export_observation, load_recovery_export,
+            };
+        }
+
         pub mod transition_transaction {
             pub use crate::bucket::lifecycle::transition_transaction::{
                 TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,
@@ -446,9 +453,12 @@ pub mod notification {
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test;
     pub use crate::services::notification_sys::{
-        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr,
-        NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
+        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, IlmRecoveryExportFleetProofToken,
+        LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
+        acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
         acquire_legacy_transition_state_reconcile_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
+        ilm_recovery_export_member_epochs_sha256, ilm_recovery_export_topology_generation,
         legacy_transition_state_reconcile_fleet_proof_matches, new_global_notification_sys,
         scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };

crates/ecstore/src/bucket/lifecycle/config_boundary.rs

@@ -41,6 +41,21 @@ where
     com::read_config(api, file).await
 }
 
+pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        >,
+{
+    com::read_config_limited_preserve_empty(api, file, max_bytes).await
+}
+
 pub(crate) async fn read_config_with_metadata<S>(api: Arc<S>, file: &str, opts: &ObjectOptions) -> Result<(Vec<u8>, ObjectInfo)>
 where
     S: ObjectIO<
@@ -56,6 +71,26 @@ where
     com::read_config_with_metadata(api, file, opts).await
 }
 
+pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+    max_bytes: usize,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        >,
+{
+    com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
+}
+
 pub(crate) async fn save_config<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()>
 where
     S: ObjectIO<

crates/ecstore/src/bucket/lifecycle/durable_namespace.rs

@@ -22,7 +22,7 @@ use super::{
     bucket_lifecycle_ops::{
         ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token,
     },
-    manual_transition_job, recovery_control, tier_delete_journal, transition_transaction,
+    manual_transition_job, recovery_control, recovery_export, tier_delete_journal, transition_transaction,
 };
 use crate::error::{Error, Result};
 use crate::services::tier::tier_probe_intent;
@@ -42,6 +42,7 @@ pub(crate) enum DurableIlmRecordKind {
     ManualTransitionTask,
     ManualTransitionWorkerResult,
     RecoveryControl,
+    RecoveryExport,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -112,8 +113,14 @@ pub(crate) const RECOVERY_CONTROL_NAMESPACE: DurableIlmNamespace = DurableIlmNam
     max_record_size: recovery_control::MAX_ILM_RECOVERY_CONTROL_SIZE,
     kind: DurableIlmRecordKind::RecoveryControl,
 };
+pub(crate) const RECOVERY_EXPORT_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
+    name: "recovery-export",
+    prefix: recovery_export::ILM_RECOVERY_EXPORT_PREFIX,
+    max_record_size: recovery_export::MAX_ILM_RECOVERY_EXPORT_SIZE,
+    kind: DurableIlmRecordKind::RecoveryExport,
+};
 
-pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 10] = [
+pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 11] = [
     TIER_DELETE_JOURNAL_NAMESPACE,
     TIER_DELETE_JOURNAL_V6_NAMESPACE,
     TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
@@ -124,6 +131,7 @@ pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 10] = [
     MANUAL_TRANSITION_TASK_NAMESPACE,
     MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
     RECOVERY_CONTROL_NAMESPACE,
+    RECOVERY_EXPORT_NAMESPACE,
 ];
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -261,6 +269,14 @@ pub(crate) enum DurableIlmRecordCheckpoint {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         owner_fence_sha256: Option<String>,
     },
+    RecoveryExport {
+        content_sha256: String,
+        source_generation_sha256: String,
+        topology_generation: String,
+        member_epochs_sha256: String,
+        creator_sha256: String,
+        retain_until_unix_nanos: i64,
+    },
 }
 
 impl DurableIlmRecordCheckpoint {
@@ -275,7 +291,8 @@ impl DurableIlmRecordCheckpoint {
             | Self::ManualTransitionScope { content_sha256, .. }
             | Self::ManualTransitionTask { content_sha256 }
             | Self::ManualTransitionWorkerResult { content_sha256 }
-            | Self::RecoveryControl { content_sha256, .. } => content_sha256,
+            | Self::RecoveryControl { content_sha256, .. }
+            | Self::RecoveryExport { content_sha256, .. } => content_sha256,
         }
     }
 
@@ -1348,6 +1365,27 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                 },
             )
         }
+        DurableIlmRecordKind::RecoveryExport => {
+            let (protocol, export_id) = recovery_export::recovery_export_id_from_record_object_name(path)?;
+            let export = recovery_export::IlmRecoveryExport::decode(&export_id, data)?;
+            let canonical = recovery_export::recovery_export_record_object_name(protocol, &export_id)?;
+            if canonical != path || export.protocol != protocol {
+                return Err(Error::other("ILM recovery export path is not canonical"));
+            }
+            let source_generation_sha256 = checkpoint_hash(&export.source_generation)?;
+            (
+                "export_id",
+                export_id,
+                DurableIlmRecordCheckpoint::RecoveryExport {
+                    content_sha256,
+                    source_generation_sha256,
+                    topology_generation: export.topology_generation,
+                    member_epochs_sha256: export.member_epochs_sha256,
+                    creator_sha256: export.creator_sha256,
+                    retain_until_unix_nanos: export.retain_until_unix_nanos,
+                },
+            )
+        }
         DurableIlmRecordKind::ManualTransitionJob => {
             let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
                 .map_err(|err| Error::other(err.to_string()))?;

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

@@ -25,6 +25,7 @@ mod object_handlers_common;
 mod object_lock_boundary;
 pub use self::core as lifecycle;
 pub mod recovery_control;
+pub mod recovery_export;
 mod replication_sink;
 pub mod rule;
 mod runtime_boundary;

crates/ecstore/src/bucket/lifecycle/recovery_control.rs

@@ -168,7 +168,7 @@ impl IlmRecoverySourceGeneration {
         Ok(generation)
     }
 
-    fn validate(&self) -> Result<()> {
+    pub(crate) fn validate(&self) -> Result<()> {
         if self.source_schema.trim().is_empty() {
             return Err(IlmRecoveryControlError::Corrupt("source schema is empty"));
         }

crates/ecstore/src/bucket/lifecycle/recovery_export.rs

@@ -0,0 +1,837 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{collections::HashSet, sync::Arc};
+
+use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
+use serde::{Deserialize, Serialize};
+
+use super::config_boundary;
+use super::recovery_control::{
+    IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryProtocol, IlmRecoverySourceCopy, IlmRecoverySourceGeneration,
+    MAX_ILM_RECOVERY_CONTROL_SIZE, ObservedIlmRecoveryControl, ObservedIlmRecoverySource, recovery_control_record_object_name,
+};
+use super::tier_delete_journal::{
+    TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA, validate_legacy_tier_delete_recovery_source,
+};
+use crate::disk::RUSTFS_META_BUCKET;
+use crate::error::{Error, Result};
+use crate::object_api::{ObjectOptions, WriteCompletion};
+use crate::services::notification_sys::{
+    acquire_ilm_recovery_export_fleet_proof, ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_member_epochs_sha256,
+    ilm_recovery_export_topology_generation,
+};
+use crate::storage_api_contracts::{list::ListOperations as _, namespace::NamespaceLocking as _, object::HTTPPreconditions};
+use crate::store::ECStore;
+
+pub const ILM_RECOVERY_EXPORT_SCHEMA: &str = "rustfs-ilm-recovery-export-v1";
+pub const ILM_RECOVERY_EXPORT_PREFIX: &str = "ilm/recovery-exports";
+pub const MAX_ILM_RECOVERY_EXPORT_SIZE: usize = 128 * 1024;
+const MAX_ILM_RECOVERY_EXPORTS: usize = 10_000;
+const MAX_ILM_RECOVERY_EXPORT_BYTES: u64 = 1024 * 1024 * 1024;
+const MAX_ACTOR_EXPORTS_PER_MINUTE: usize = 10;
+const MAX_CLUSTER_EXPORTS_PER_MINUTE: usize = 100;
+const EXPORT_RETENTION_NANOS: i64 = 90 * 24 * 60 * 60 * 1_000_000_000;
+const EXPORT_ADMISSION_LOCK: &str = "ilm/recovery-admission/export.lock";
+const MAX_LEGACY_TIER_DELETE_SOURCE_SIZE: usize = 64 * 1024;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct IlmRecoveryExportObservation {
+    pub control_id: String,
+    pub protocol: IlmRecoveryProtocol,
+    pub control_etag: String,
+    pub control_revision: u64,
+    pub classification: IlmRecoveryClassification,
+    pub canonical_source_path: String,
+    pub source_generation: IlmRecoverySourceGeneration,
+    pub topology_generation: String,
+    pub member_epochs_sha256: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct IlmRecoveryExport {
+    pub export_id: String,
+    pub control_id: String,
+    pub protocol: IlmRecoveryProtocol,
+    pub control_etag: String,
+    pub control_revision: u64,
+    pub classification: IlmRecoveryClassification,
+    pub canonical_source_path: String,
+    pub source_generation: IlmRecoverySourceGeneration,
+    pub topology_generation: String,
+    pub member_epochs_sha256: String,
+    pub creator_sha256: String,
+    pub created_at_unix_nanos: i64,
+    pub retain_until_unix_nanos: i64,
+    pub source_bytes_base64: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PersistedIlmRecoveryExport {
+    schema: String,
+    content_sha256: String,
+    export: IlmRecoveryExport,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct IlmRecoveryExportCreated {
+    pub export_id: String,
+    pub content_sha256: String,
+    pub encoded: Vec<u8>,
+    pub replayed: bool,
+}
+
+impl IlmRecoveryExport {
+    fn validate(&self) -> Result<()> {
+        self.source_generation.validate().map_err(Error::other)?;
+        validate_sha256(&self.export_id, "ILM recovery export ID is invalid")?;
+        validate_sha256(&self.control_id, "ILM recovery export control ID is invalid")?;
+        validate_sha256(&self.topology_generation, "ILM recovery export topology generation is invalid")?;
+        validate_sha256(&self.member_epochs_sha256, "ILM recovery export member epoch digest is invalid")?;
+        validate_sha256(&self.creator_sha256, "ILM recovery export creator digest is invalid")?;
+        if self.protocol != IlmRecoveryProtocol::TierDeleteJournal
+            || self.classification != IlmRecoveryClassification::RetainedAmbiguous
+            || !is_legacy_export_schema(&self.source_generation.source_schema)
+        {
+            return Err(Error::other("ILM recovery export source is not an exportable legacy journal"));
+        }
+        if self.control_etag.trim().is_empty() || self.control_revision == 0 {
+            return Err(Error::other("ILM recovery export control generation is invalid"));
+        }
+        if self.canonical_source_path.is_empty()
+            || self.canonical_source_path.starts_with('/')
+            || self.canonical_source_path.ends_with('/')
+            || self.canonical_source_path.split('/').any(str::is_empty)
+        {
+            return Err(Error::other("ILM recovery export source path is invalid"));
+        }
+        if self.created_at_unix_nanos <= 0
+            || self.retain_until_unix_nanos < self.created_at_unix_nanos.saturating_add(EXPORT_RETENTION_NANOS)
+        {
+            return Err(Error::other("ILM recovery export retention is invalid"));
+        }
+        let source = base64_simd::STANDARD
+            .decode_to_vec(self.source_bytes_base64.as_bytes())
+            .map_err(|_| Error::other("ILM recovery export source encoding is invalid"))?;
+        validate_legacy_tier_delete_recovery_source(&self.canonical_source_path, &self.source_generation.source_schema, &source)?;
+        let encoded_len = u64::try_from(source.len()).map_err(|_| Error::other("ILM recovery export source length overflow"))?;
+        if source.is_empty()
+            || source.len() > MAX_LEGACY_TIER_DELETE_SOURCE_SIZE
+            || hex_sha256(&source, ToOwned::to_owned) != self.source_generation.content_sha256
+            || self.source_generation.copies.iter().any(|copy| {
+                copy.canonical_path != self.canonical_source_path
+                    || copy.etag != self.source_generation.source_etag
+                    || copy.content_sha256 != self.source_generation.content_sha256
+                    || copy.encoded_len != encoded_len
+            })
+        {
+            return Err(Error::other("ILM recovery export source bytes do not match the observed generation"));
+        }
+        if export_id(&self.control_id, &self.source_generation)? != self.export_id {
+            return Err(Error::other("ILM recovery export ID does not match its source generation"));
+        }
+        Ok(())
+    }
+
+    pub fn encode(&self) -> Result<Vec<u8>> {
+        self.validate()?;
+        let export_bytes = serde_json::to_vec(self).map_err(Error::other)?;
+        let persisted = PersistedIlmRecoveryExport {
+            schema: ILM_RECOVERY_EXPORT_SCHEMA.to_string(),
+            content_sha256: hex_sha256(&export_bytes, ToOwned::to_owned),
+            export: self.clone(),
+        };
+        let encoded = serde_json::to_vec(&persisted).map_err(Error::other)?;
+        if encoded.len() > MAX_ILM_RECOVERY_EXPORT_SIZE {
+            return Err(Error::other("encoded ILM recovery export exceeds maximum size"));
+        }
+        Ok(encoded)
+    }
+
+    pub fn decode(expected_export_id: &str, data: &[u8]) -> Result<Self> {
+        validate_sha256(expected_export_id, "ILM recovery export ID is invalid")?;
+        if data.len() > MAX_ILM_RECOVERY_EXPORT_SIZE {
+            return Err(Error::other("encoded ILM recovery export exceeds maximum size"));
+        }
+        let persisted: PersistedIlmRecoveryExport = serde_json::from_slice(data).map_err(Error::other)?;
+        if persisted.schema != ILM_RECOVERY_EXPORT_SCHEMA {
+            return Err(Error::other("ILM recovery export schema is unsupported"));
+        }
+        validate_sha256(&persisted.content_sha256, "ILM recovery export checksum is invalid")?;
+        let export_bytes = serde_json::to_vec(&persisted.export).map_err(Error::other)?;
+        if hex_sha256(&export_bytes, ToOwned::to_owned) != persisted.content_sha256 {
+            return Err(Error::other("ILM recovery export checksum mismatch"));
+        }
+        persisted.export.validate()?;
+        if persisted.export.export_id != expected_export_id {
+            return Err(Error::other("ILM recovery export ID does not match record key"));
+        }
+        Ok(persisted.export)
+    }
+}
+
+pub fn recovery_export_record_object_name(protocol: IlmRecoveryProtocol, export_id: &str) -> Result<String> {
+    validate_sha256(export_id, "ILM recovery export ID is invalid")?;
+    Ok(format!(
+        "{}/{}/{}/{}/{}.json",
+        ILM_RECOVERY_EXPORT_PREFIX,
+        protocol.as_str(),
+        &export_id[..2],
+        &export_id[2..4],
+        export_id
+    ))
+}
+
+pub fn recovery_export_id_from_record_object_name(object: &str) -> Result<(IlmRecoveryProtocol, String)> {
+    let suffix = object
+        .strip_prefix(ILM_RECOVERY_EXPORT_PREFIX)
+        .and_then(|suffix| suffix.strip_prefix('/'))
+        .ok_or_else(|| Error::other("ILM recovery export path has wrong prefix"))?;
+    let mut parts = suffix.split('/');
+    let protocol = match parts.next() {
+        Some("tier_delete_journal") => IlmRecoveryProtocol::TierDeleteJournal,
+        _ => return Err(Error::other("ILM recovery export protocol is invalid")),
+    };
+    let shard_a = parts
+        .next()
+        .ok_or_else(|| Error::other("ILM recovery export path is incomplete"))?;
+    let shard_b = parts
+        .next()
+        .ok_or_else(|| Error::other("ILM recovery export path is incomplete"))?;
+    let export_id = parts
+        .next()
+        .and_then(|name| name.strip_suffix(".json"))
+        .ok_or_else(|| Error::other("ILM recovery export suffix is invalid"))?;
+    if parts.next().is_some() {
+        return Err(Error::other("ILM recovery export path is not canonical"));
+    }
+    validate_sha256(export_id, "ILM recovery export ID is invalid")?;
+    if shard_a != &export_id[..2] || shard_b != &export_id[2..4] {
+        return Err(Error::other("ILM recovery export shard does not match export ID"));
+    }
+    Ok((protocol, export_id.to_string()))
+}
+
+pub async fn inspect_recovery_export_observation(api: Arc<ECStore>, control_id: &str) -> Result<IlmRecoveryExportObservation> {
+    let proof = acquire_ilm_recovery_export_fleet_proof()
+        .await
+        .ok_or_else(|| Error::other("ILM recovery export fleet proof is unavailable"))?;
+    let observed_control = load_exportable_control(api.clone(), control_id).await?;
+    let observed_source = observe_export_source(
+        api,
+        &observed_control.control.identity.canonical_source_path,
+        &observed_control.control.observed_source_generation.source_schema,
+    )
+    .await?;
+    if !observed_source.is_consistent()
+        || observed_source.generation != observed_control.control.observed_source_generation
+        || !ilm_recovery_export_fleet_proof_matches(&proof).await
+    {
+        return Err(Error::other("ILM recovery export observation changed or is incomplete"));
+    }
+    Ok(IlmRecoveryExportObservation {
+        control_id: control_id.to_string(),
+        protocol: observed_control.control.identity.protocol,
+        control_etag: observed_control.etag,
+        control_revision: observed_control.control.revision,
+        classification: observed_control.control.classification,
+        canonical_source_path: observed_control.control.identity.canonical_source_path,
+        source_generation: observed_source.generation,
+        topology_generation: ilm_recovery_export_topology_generation(&proof),
+        member_epochs_sha256: ilm_recovery_export_member_epochs_sha256(&proof),
+    })
+}
+
+pub async fn create_recovery_export(
+    api: Arc<ECStore>,
+    observation: &IlmRecoveryExportObservation,
+    creator_sha256: &str,
+) -> Result<IlmRecoveryExportCreated> {
+    validate_sha256(creator_sha256, "ILM recovery export creator digest is invalid")?;
+    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, EXPORT_ADMISSION_LOCK).await?;
+    let admission_guard = lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
+
+    let proof = acquire_ilm_recovery_export_fleet_proof()
+        .await
+        .ok_or_else(|| Error::other("ILM recovery export fleet proof is unavailable"))?;
+    if ilm_recovery_export_topology_generation(&proof) != observation.topology_generation
+        || ilm_recovery_export_member_epochs_sha256(&proof) != observation.member_epochs_sha256
+    {
+        return Err(Error::PreconditionFailed);
+    }
+    let control_object = recovery_control_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, &observation.control_id)
+        .map_err(Error::other)?;
+    let control_lock = api.new_ns_lock(RUSTFS_META_BUCKET, &control_object).await?;
+    let control_guard = control_lock
+        .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await?;
+    let source_lock = api
+        .new_ns_lock(RUSTFS_META_BUCKET, &observation.canonical_source_path)
+        .await?;
+    let source_guard = source_lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
+    let locks_current = || !admission_guard.is_lock_lost() && !control_guard.is_lock_lost() && !source_guard.is_lock_lost();
+    let (current, current_source_bytes) = current_observation_under_proof_no_lock(api.clone(), observation, &proof).await?;
+    if &current != observation || !locks_current() {
+        return Err(Error::PreconditionFailed);
+    }
+    let current_source_base64 = base64_simd::STANDARD.encode_to_string(current_source_bytes);
+    let candidate_export_id = export_id(&current.control_id, &current.source_generation)?;
+    let object = recovery_export_record_object_name(current.protocol, &candidate_export_id)?;
+    match load_recovery_export_decoded(api.clone(), &candidate_export_id).await {
+        Ok((existing, export)) if export_matches_observation(&export, observation) => {
+            if !locks_current() || !ilm_recovery_export_fleet_proof_matches(&proof).await {
+                return Err(Error::PreconditionFailed);
+            }
+            api.record_durable_ilm_decommission_progress(&object, &existing.encoded)
+                .await?;
+            if !locks_current() {
+                return Err(Error::PreconditionFailed);
+            }
+            return Ok(existing.with_replayed());
+        }
+        Ok(_) => return Err(Error::PreconditionFailed),
+        Err(Error::ConfigNotFound) => {}
+        Err(err) => return Err(err),
+    }
+    let inventory = collect_export_inventory(api.clone()).await?;
+    if !locks_current() || !ilm_recovery_export_fleet_proof_matches(&proof).await {
+        return Err(Error::PreconditionFailed);
+    }
+    let created_at_unix_nanos = now_unix_nanos()?;
+    let export = build_export_from_source(&current, creator_sha256, created_at_unix_nanos, &current_source_base64)?;
+    let encoded = export.encode()?;
+    inventory.check(creator_sha256, encoded.len(), created_at_unix_nanos)?;
+
+    let mut write_options = ObjectOptions {
+        max_parity: true,
+        write_completion: WriteCompletion::TailDrained,
+        http_preconditions: Some(HTTPPreconditions {
+            if_none_match: Some("*".to_string()),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    write_options.add_namespace_lock_guard(&admission_guard);
+    write_options.add_namespace_lock_guard(&control_guard);
+    write_options.add_namespace_lock_guard(&source_guard);
+    if !locks_current() {
+        return Err(Error::PreconditionFailed);
+    }
+    let write_result = config_boundary::save_config_with_opts(api.clone(), &object, encoded.clone(), &write_options).await;
+    let stored = match load_recovery_export(api.clone(), &export.export_id).await {
+        Ok(stored) if stored.encoded == encoded => stored,
+        Ok(_) => return Err(Error::PreconditionFailed),
+        Err(read_err) => return Err(write_result.err().unwrap_or(read_err)),
+    };
+    if !locks_current() || !ilm_recovery_export_fleet_proof_matches(&proof).await {
+        return Err(Error::PreconditionFailed);
+    }
+    api.record_durable_ilm_decommission_progress(&object, &encoded).await?;
+    if !locks_current() {
+        return Err(Error::PreconditionFailed);
+    }
+    Ok(stored)
+}
+
+pub async fn load_recovery_export(api: Arc<ECStore>, export_id: &str) -> Result<IlmRecoveryExportCreated> {
+    let (created, _) = load_recovery_export_decoded(api, export_id).await?;
+    Ok(created)
+}
+
+async fn load_recovery_export_decoded(
+    api: Arc<ECStore>,
+    export_id: &str,
+) -> Result<(IlmRecoveryExportCreated, IlmRecoveryExport)> {
+    let object = recovery_export_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, export_id)?;
+    let encoded = config_boundary::read_config_limited_preserve_empty(api, &object, MAX_ILM_RECOVERY_EXPORT_SIZE).await?;
+    let export = IlmRecoveryExport::decode(export_id, &encoded)?;
+    let content_sha256 = hex_sha256(&encoded, ToOwned::to_owned);
+    Ok((
+        IlmRecoveryExportCreated {
+            export_id: export.export_id.clone(),
+            content_sha256,
+            encoded,
+            replayed: false,
+        },
+        export,
+    ))
+}
+
+impl IlmRecoveryExportCreated {
+    fn with_replayed(mut self) -> Self {
+        self.replayed = true;
+        self
+    }
+}
+
+async fn load_exportable_control(api: Arc<ECStore>, control_id: &str) -> Result<ObservedIlmRecoveryControl> {
+    load_exportable_control_with_options(api, control_id, &ObjectOptions::default()).await
+}
+
+async fn load_exportable_control_no_lock(api: Arc<ECStore>, control_id: &str) -> Result<ObservedIlmRecoveryControl> {
+    load_exportable_control_with_options(
+        api,
+        control_id,
+        &ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        },
+    )
+    .await
+}
+
+async fn load_exportable_control_with_options(
+    api: Arc<ECStore>,
+    control_id: &str,
+    options: &ObjectOptions,
+) -> Result<ObservedIlmRecoveryControl> {
+    let object = recovery_control_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, control_id).map_err(Error::other)?;
+    let (data, metadata) =
+        config_boundary::read_config_limited_preserve_empty_with_metadata(api, &object, options, MAX_ILM_RECOVERY_CONTROL_SIZE)
+            .await?;
+    let etag = metadata
+        .etag
+        .filter(|etag| !etag.trim().is_empty())
+        .ok_or_else(|| Error::other("ILM recovery control is missing an ETag"))?;
+    let control = IlmRecoveryControl::decode(control_id, &data).map_err(Error::other)?;
+    if control.identity.protocol != IlmRecoveryProtocol::TierDeleteJournal
+        || control.classification != IlmRecoveryClassification::RetainedAmbiguous
+        || !is_legacy_export_schema(&control.observed_source_generation.source_schema)
+    {
+        return Err(Error::other("ILM recovery control is not exportable"));
+    }
+    Ok(ObservedIlmRecoveryControl { control, etag })
+}
+
+async fn current_observation_under_proof_no_lock(
+    api: Arc<ECStore>,
+    expected: &IlmRecoveryExportObservation,
+    proof: &crate::services::notification_sys::IlmRecoveryExportFleetProofToken,
+) -> Result<(IlmRecoveryExportObservation, Vec<u8>)> {
+    let observed_control = load_exportable_control_no_lock(api.clone(), &expected.control_id).await?;
+    let observed_source = observe_export_source_no_lock(
+        api,
+        &observed_control.control.identity.canonical_source_path,
+        &observed_control.control.observed_source_generation.source_schema,
+    )
+    .await?;
+    let source_bytes = observed_source
+        .canonical_data
+        .clone()
+        .ok_or_else(|| Error::other("ILM recovery export source copies diverge"))?;
+    if !observed_source.is_consistent()
+        || observed_source.generation != observed_control.control.observed_source_generation
+        || !ilm_recovery_export_fleet_proof_matches(proof).await
+    {
+        return Err(Error::PreconditionFailed);
+    }
+    Ok((
+        IlmRecoveryExportObservation {
+            control_id: expected.control_id.clone(),
+            protocol: observed_control.control.identity.protocol,
+            control_etag: observed_control.etag,
+            control_revision: observed_control.control.revision,
+            classification: observed_control.control.classification,
+            canonical_source_path: observed_control.control.identity.canonical_source_path,
+            source_generation: observed_source.generation,
+            topology_generation: ilm_recovery_export_topology_generation(proof),
+            member_epochs_sha256: ilm_recovery_export_member_epochs_sha256(proof),
+        },
+        source_bytes,
+    ))
+}
+
+async fn observe_export_source(
+    api: Arc<ECStore>,
+    canonical_path: &str,
+    source_schema: &str,
+) -> Result<ObservedIlmRecoverySource> {
+    if canonical_path.is_empty()
+        || canonical_path.starts_with('/')
+        || canonical_path.ends_with('/')
+        || canonical_path.split('/').any(str::is_empty)
+        || !is_legacy_export_schema(source_schema)
+    {
+        return Err(Error::other("ILM recovery export source identity is invalid"));
+    }
+    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, canonical_path).await?;
+    let _guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
+    observe_export_source_no_lock(api, canonical_path, source_schema).await
+}
+
+async fn observe_export_source_no_lock(
+    api: Arc<ECStore>,
+    canonical_path: &str,
+    source_schema: &str,
+) -> Result<ObservedIlmRecoverySource> {
+    let mut copies = Vec::new();
+    let mut canonical: Option<(String, String, Vec<u8>)> = None;
+    let mut consistent = true;
+    for set in api.all_set_disks() {
+        let authority = format!("pool-{}/set-{}", set.pool_index, set.set_index);
+        let result = config_boundary::read_config_limited_preserve_empty_with_metadata(
+            set,
+            canonical_path,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+            MAX_LEGACY_TIER_DELETE_SOURCE_SIZE,
+        )
+        .await;
+        match result {
+            Ok((data, metadata)) => {
+                if data.is_empty() || data.len() > MAX_LEGACY_TIER_DELETE_SOURCE_SIZE {
+                    return Err(Error::other("ILM recovery export source exceeds its protocol size limit"));
+                }
+                validate_legacy_tier_delete_recovery_source(canonical_path, source_schema, &data)?;
+                let etag = metadata
+                    .etag
+                    .filter(|etag| !etag.trim().is_empty())
+                    .ok_or_else(|| Error::other("ILM recovery export source copy is missing an ETag"))?;
+                let content_sha256 = hex_sha256(&data, ToOwned::to_owned);
+                let encoded_len =
+                    u64::try_from(data.len()).map_err(|_| Error::other("ILM recovery export source length does not fit u64"))?;
+                copies.push(IlmRecoverySourceCopy {
+                    authority,
+                    canonical_path: canonical_path.to_string(),
+                    etag: etag.clone(),
+                    encoded_len,
+                    content_sha256: content_sha256.clone(),
+                });
+                match canonical.as_ref() {
+                    Some((first_etag, first_digest, first_data)) => {
+                        consistent &= first_etag == &etag && first_digest == &content_sha256 && first_data == &data;
+                    }
+                    None => canonical = Some((etag, content_sha256, data)),
+                }
+            }
+            Err(err) if export_source_is_missing(&err) => {}
+            Err(err) => return Err(err),
+        }
+    }
+    let Some((source_etag, content_sha256, source_bytes)) = canonical else {
+        return Err(Error::ConfigNotFound);
+    };
+    let generation =
+        IlmRecoverySourceGeneration::new(source_schema, source_etag, content_sha256, copies).map_err(Error::other)?;
+    Ok(ObservedIlmRecoverySource {
+        generation,
+        canonical_data: consistent.then_some(source_bytes),
+    })
+}
+
+fn export_source_is_missing(err: &Error) -> bool {
+    matches!(
+        err,
+        Error::ConfigNotFound | Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::VersionNotFound(_, _, _)
+    )
+}
+
+fn build_export_from_source(
+    observation: &IlmRecoveryExportObservation,
+    creator_sha256: &str,
+    created_at_unix_nanos: i64,
+    source_bytes_base64: &str,
+) -> Result<IlmRecoveryExport> {
+    let retain_until_unix_nanos = created_at_unix_nanos
+        .checked_add(EXPORT_RETENTION_NANOS)
+        .ok_or_else(|| Error::other("ILM recovery export retention timestamp overflow"))?;
+    let export = IlmRecoveryExport {
+        export_id: export_id(&observation.control_id, &observation.source_generation)?,
+        control_id: observation.control_id.clone(),
+        protocol: observation.protocol,
+        control_etag: observation.control_etag.clone(),
+        control_revision: observation.control_revision,
+        classification: observation.classification,
+        canonical_source_path: observation.canonical_source_path.clone(),
+        source_generation: observation.source_generation.clone(),
+        topology_generation: observation.topology_generation.clone(),
+        member_epochs_sha256: observation.member_epochs_sha256.clone(),
+        creator_sha256: creator_sha256.to_string(),
+        created_at_unix_nanos,
+        retain_until_unix_nanos,
+        source_bytes_base64: source_bytes_base64.to_string(),
+    };
+    export.validate()?;
+    Ok(export)
+}
+
+fn export_id(control_id: &str, generation: &IlmRecoverySourceGeneration) -> Result<String> {
+    validate_sha256(control_id, "ILM recovery export control ID is invalid")?;
+    validate_sha256(&generation.content_sha256, "ILM recovery export source checksum is invalid")?;
+    validate_sha256(&generation.copy_set_sha256, "ILM recovery export copy-set checksum is invalid")?;
+    let mut data = Vec::new();
+    for part in [control_id, &generation.content_sha256, &generation.copy_set_sha256] {
+        data.extend_from_slice(&(part.len() as u64).to_be_bytes());
+        data.extend_from_slice(part.as_bytes());
+    }
+    Ok(hex_sha256(&data, ToOwned::to_owned))
+}
+
+fn export_matches_observation(export: &IlmRecoveryExport, observation: &IlmRecoveryExportObservation) -> bool {
+    export.control_id == observation.control_id
+        && export.protocol == observation.protocol
+        && export.classification == observation.classification
+        && export.canonical_source_path == observation.canonical_source_path
+        && export.source_generation == observation.source_generation
+}
+
+#[derive(Debug, Default)]
+struct IlmRecoveryExportInventory {
+    count: usize,
+    bytes: u64,
+    creations: Vec<(i64, String)>,
+}
+
+impl IlmRecoveryExportInventory {
+    fn check(&self, creator_sha256: &str, candidate_len: usize, now: i64) -> Result<()> {
+        let recent_after = now.saturating_sub(60 * 1_000_000_000);
+        let cluster_recent = self
+            .creations
+            .iter()
+            .filter(|(created_at, _)| *created_at > recent_after)
+            .count();
+        let actor_recent = self
+            .creations
+            .iter()
+            .filter(|(created_at, creator)| *created_at > recent_after && creator == creator_sha256)
+            .count();
+        check_export_admission(self.count, self.bytes, actor_recent, cluster_recent, candidate_len)
+    }
+}
+
+async fn collect_export_inventory(api: Arc<ECStore>) -> Result<IlmRecoveryExportInventory> {
+    let mut marker = None;
+    let mut seen_markers = HashSet::new();
+    let mut inventory = IlmRecoveryExportInventory::default();
+    loop {
+        let page = api
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                &format!("{ILM_RECOVERY_EXPORT_PREFIX}/"),
+                marker.clone(),
+                None,
+                1_000,
+                false,
+                None,
+                false,
+            )
+            .await?;
+        for object in page.objects {
+            let (_, export_id) = recovery_export_id_from_record_object_name(&object.name)?;
+            let (stored, export) = load_recovery_export_decoded(api.clone(), &export_id).await?;
+            inventory.count = inventory
+                .count
+                .checked_add(1)
+                .ok_or_else(|| Error::other("ILM recovery export count overflow"))?;
+            inventory.bytes = inventory
+                .bytes
+                .checked_add(u64::try_from(stored.encoded.len()).map_err(|_| Error::other("ILM recovery export size overflow"))?)
+                .ok_or_else(|| Error::other("ILM recovery export byte total overflow"))?;
+            inventory
+                .creations
+                .push((export.created_at_unix_nanos, export.creator_sha256));
+        }
+        if !page.is_truncated {
+            break;
+        }
+        let next = page
+            .next_continuation_token
+            .ok_or_else(|| Error::other("ILM recovery export inventory omitted its continuation marker"))?;
+        marker = Some(record_export_inventory_marker(&mut seen_markers, next)?);
+    }
+    Ok(inventory)
+}
+
+fn record_export_inventory_marker(seen_markers: &mut HashSet<String>, next: String) -> Result<String> {
+    if !seen_markers.insert(next.clone()) {
+        return Err(Error::other("ILM recovery export inventory repeated its continuation marker"));
+    }
+    Ok(next)
+}
+
+fn check_export_admission(
+    count: usize,
+    bytes: u64,
+    actor_recent: usize,
+    cluster_recent: usize,
+    candidate_len: usize,
+) -> Result<()> {
+    let candidate_len = u64::try_from(candidate_len).map_err(|_| Error::other("ILM recovery export size does not fit u64"))?;
+    if count >= MAX_ILM_RECOVERY_EXPORTS
+        || bytes
+            .checked_add(candidate_len)
+            .is_none_or(|total| total > MAX_ILM_RECOVERY_EXPORT_BYTES)
+        || actor_recent >= MAX_ACTOR_EXPORTS_PER_MINUTE
+        || cluster_recent >= MAX_CLUSTER_EXPORTS_PER_MINUTE
+    {
+        return Err(Error::SlowDown);
+    }
+    Ok(())
+}
+
+fn is_legacy_export_schema(schema: &str) -> bool {
+    matches!(schema, TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA | TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA)
+}
+
+fn validate_sha256(value: &str, message: &'static str) -> Result<()> {
+    if !is_sha256_checksum(value)
+        || value
+            .bytes()
+            .any(|byte| byte.is_ascii_hexdigit() && byte.is_ascii_uppercase())
+    {
+        return Err(Error::other(message));
+    }
+    Ok(())
+}
+
+fn now_unix_nanos() -> Result<i64> {
+    i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos())
+        .map_err(|_| Error::other("ILM recovery export timestamp does not fit i64"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bucket::lifecycle::recovery_control::IlmRecoverySourceCopy;
+
+    const PINNED_V1_EXPORT: &[u8] = br#"{"schema":"rustfs-ilm-recovery-export-v1","content_sha256":"3dfb3ec3892256e909de1211c1a963ca7008963ff32b3a869f7161a7b9b44028","export":{"export_id":"2b78e7a825bfc2edbf7f773d0b6ed3bf93e360ff1702d73a449109c11bfaa105","control_id":"0fcd568a5cb9bdb4677b69354b11ee415af8f784519cff3da49a26f84eaee7f2","protocol":"tier_delete_journal","control_etag":"control-etag","control_revision":1,"classification":"retained_ambiguous","canonical_source_path":"ilm/tier-delete-journal/872072554f66ab326f10ce7adbae11422b7a4b0663aa7112d6061a8f6ed41b94.json","source_generation":{"source_schema":"rustfs-tier-delete-journal-v1","source_etag":"etag-a","content_sha256":"0e0b010ebdeeb7b41473fe8575e989d6bb1303c0ca551dd984e9400f0ae306bd","copy_set_sha256":"5a7406115b6c3923ffe79dcd1f43ccae7beed786e557163f019dd10ec409a653","copies":[{"authority":"pool-0/set-0","canonical_path":"ilm/tier-delete-journal/872072554f66ab326f10ce7adbae11422b7a4b0663aa7112d6061a8f6ed41b94.json","etag":"etag-a","encoded_len":81,"content_sha256":"0e0b010ebdeeb7b41473fe8575e989d6bb1303c0ca551dd984e9400f0ae306bd"}]},"topology_generation":"e6e2b826e31fca5c36125c48f130dcb6f961e698ff8a8776a1f290cf0892e8e6","member_epochs_sha256":"612dd8a861161819a4ad8f6f3e2a0567602877c043a2353ca933a13c78dc0ed4","creator_sha256":"50c9c4aeb40b5b206b6d98f516f8b8c0efd29ce2e56a76b345fb9240c225a1b7","created_at_unix_nanos":1000000000,"retain_until_unix_nanos":7776001000000000,"source_bytes_base64":"eyJ2ZXJzaW9uIjoxLCJvYmpfbmFtZSI6ImxlZ2FjeS9yZW1vdGUiLCJ2ZXJzaW9uX2lkIjoib3BhcXVlIiwidGllcl9uYW1lIjoiV0FSTSJ9"}}"#;
+
+    fn legacy_source() -> Vec<u8> {
+        br#"{"version":1,"obj_name":"legacy/remote","version_id":"opaque","tier_name":"WARM"}"#.to_vec()
+    }
+
+    fn observation() -> IlmRecoveryExportObservation {
+        let source = legacy_source();
+        let source_path = super::super::tier_delete_journal::tier_delete_journal_object_name(
+            &super::super::tier_delete_journal::decode_tier_delete_journal_entry(&source).expect("legacy fixture should decode"),
+        );
+        let source_sha256 = hex_sha256(&source, ToOwned::to_owned);
+        let generation = IlmRecoverySourceGeneration::new(
+            TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA,
+            "etag-a",
+            source_sha256.clone(),
+            vec![IlmRecoverySourceCopy {
+                authority: "pool-0/set-0".to_string(),
+                canonical_path: source_path.clone(),
+                etag: "etag-a".to_string(),
+                encoded_len: source.len() as u64,
+                content_sha256: source_sha256,
+            }],
+        )
+        .expect("generation should be valid");
+        IlmRecoveryExportObservation {
+            control_id: hex_sha256(b"control", ToOwned::to_owned),
+            protocol: IlmRecoveryProtocol::TierDeleteJournal,
+            control_etag: "control-etag".to_string(),
+            control_revision: 1,
+            classification: IlmRecoveryClassification::RetainedAmbiguous,
+            canonical_source_path: source_path,
+            source_generation: generation,
+            topology_generation: hex_sha256(b"topology", ToOwned::to_owned),
+            member_epochs_sha256: hex_sha256(b"epochs", ToOwned::to_owned),
+        }
+    }
+
+    #[test]
+    fn recovery_export_round_trip_is_strict_and_deterministic() {
+        let observed = observation();
+        let creator = hex_sha256(b"actor", ToOwned::to_owned);
+        let export = build_export_from_source(
+            &observed,
+            &creator,
+            1_000_000_000,
+            &base64_simd::STANDARD.encode_to_string(legacy_source()),
+        )
+        .expect("export should be valid");
+        assert_eq!(export.export_id, export_id(&observed.control_id, &observed.source_generation).unwrap());
+        let encoded = export.encode().expect("export should encode");
+        assert_eq!(encoded, PINNED_V1_EXPORT, "v1 export wire format must remain pinned");
+        assert_eq!(IlmRecoveryExport::decode(&export.export_id, &encoded).unwrap(), export);
+        assert_eq!(
+            IlmRecoveryExport::decode("2b78e7a825bfc2edbf7f773d0b6ed3bf93e360ff1702d73a449109c11bfaa105", PINNED_V1_EXPORT)
+                .unwrap(),
+            export,
+        );
+
+        let path = recovery_export_record_object_name(export.protocol, &export.export_id).unwrap();
+        let durable = super::super::durable_namespace::validate_durable_ilm_record(&path, &encoded)
+            .expect("export should be registered as a durable ILM record");
+        assert_eq!(durable.namespace, "recovery-export");
+        assert_eq!(durable.id_kind, "export_id");
+        assert_eq!(durable.id, export.export_id);
+
+        let mut wrong_source = export.clone();
+        wrong_source.source_bytes_base64 = base64_simd::STANDARD.encode_to_string(b"changed");
+        assert!(wrong_source.encode().is_err());
+
+        let mut persisted: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
+        persisted["unknown"] = serde_json::json!(true);
+        assert!(IlmRecoveryExport::decode(&export.export_id, &serde_json::to_vec(&persisted).unwrap()).is_err());
+    }
+
+    #[test]
+    fn export_inventory_rejects_non_adjacent_continuation_cycles() {
+        let mut seen = HashSet::new();
+        assert_eq!(record_export_inventory_marker(&mut seen, "a".to_string()).unwrap(), "a");
+        assert_eq!(record_export_inventory_marker(&mut seen, "b".to_string()).unwrap(), "b");
+        record_export_inventory_marker(&mut seen, "a".to_string())
+            .expect_err("a non-adjacent continuation marker cycle must fail closed");
+    }
+
+    #[test]
+    fn recovery_export_path_rejects_noncanonical_shards() {
+        let id = hex_sha256(b"export", ToOwned::to_owned);
+        let path = recovery_export_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, &id).unwrap();
+        assert_eq!(recovery_export_id_from_record_object_name(&path).unwrap().1, id);
+        let wrong_shard = path.replacen(&format!("/{}/", &id[..2]), "/zz/", 1);
+        assert!(recovery_export_id_from_record_object_name(&wrong_shard).is_err());
+    }
+
+    #[test]
+    fn canonical_replay_survives_fleet_rotation_but_not_source_change() {
+        let observed = observation();
+        let creator = hex_sha256(b"actor", ToOwned::to_owned);
+        let export = build_export_from_source(
+            &observed,
+            &creator,
+            1_000_000_000,
+            &base64_simd::STANDARD.encode_to_string(legacy_source()),
+        )
+        .unwrap();
+        let mut rotated = observed.clone();
+        rotated.control_etag = "new-control-etag".to_string();
+        rotated.control_revision += 1;
+        rotated.topology_generation = hex_sha256(b"new-topology", ToOwned::to_owned);
+        rotated.member_epochs_sha256 = hex_sha256(b"new-members", ToOwned::to_owned);
+        assert!(export_matches_observation(&export, &rotated));
+
+        rotated.source_generation.content_sha256 = hex_sha256(b"changed", ToOwned::to_owned);
+        assert!(!export_matches_observation(&export, &rotated));
+    }
+
+    #[test]
+    fn export_admission_enforces_exact_count_byte_and_rate_boundaries() {
+        assert!(check_export_admission(9_999, MAX_ILM_RECOVERY_EXPORT_BYTES - 1, 9, 99, 1).is_ok());
+        assert!(check_export_admission(10_000, 0, 0, 0, 1).is_err());
+        assert!(check_export_admission(0, MAX_ILM_RECOVERY_EXPORT_BYTES, 0, 0, 1).is_err());
+        assert!(check_export_admission(0, 0, 10, 0, 1).is_err());
+        assert!(check_export_admission(0, 0, 0, 100, 1).is_err());
+    }
+}

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -82,8 +82,8 @@ const TIER_DELETE_DISPATCH_MEMBER_DELETE_CONCURRENCY: usize = 32;
 const TIER_DELETE_DISPATCH_PREPARE_CONCURRENCY: usize = 16;
 const TIER_DELETE_DISPATCH_CAS_CONCURRENCY: usize = 32;
 const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
-const TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v1";
-const TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v2";
+pub(crate) const TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v1";
+pub(crate) const TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v2";
 const TIER_DELETE_JOURNAL_UNKNOWN_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-unknown";
 const TIER_DELETE_JOURNAL_V1_RECOVERY_CLASS: &str = "tier_delete_journal_v1";
 const TIER_DELETE_JOURNAL_V2_RECOVERY_CLASS: &str = "tier_delete_journal_v2";
@@ -884,6 +884,23 @@ struct PersistedTierDeleteJournalEntry {
 }
 
 impl PersistedTierDeleteJournalEntry {
+    fn validate_legacy_recovery_shape(&self) -> Result<()> {
+        let has_later_version_fields = self.version_id_exact.is_some()
+            || self.version_state.is_some()
+            || self.state.is_some()
+            || self.source.is_some()
+            || self.dispatch.is_some();
+        match self.version {
+            1 if self.backend_identity.is_none() && !has_later_version_fields => Ok(()),
+            TIER_DELETE_JOURNAL_VERSION if self.backend_identity.is_some() && !has_later_version_fields => Ok(()),
+            1 => Err(Error::other("tier delete journal v1 entry contains fields from a later version")),
+            TIER_DELETE_JOURNAL_VERSION => Err(Error::other(
+                "tier delete journal v2 entry is missing its identity or contains fields from a later version",
+            )),
+            _ => Err(Error::other("tier delete journal is not an exportable legacy version")),
+        }
+    }
+
     fn from_jentry(je: &Jentry) -> Result<Self> {
         validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?;
         let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown;
@@ -5539,6 +5556,23 @@ fn legacy_tier_delete_recovery_descriptor(entry: &Jentry) -> Option<(&'static st
     }
 }
 
+pub(crate) fn validate_legacy_tier_delete_recovery_source(object_name: &str, source_schema: &str, data: &[u8]) -> Result<()> {
+    if canonical_legacy_tier_delete_journal_identity(object_name).is_none() {
+        return Err(Error::other("legacy tier delete journal path is not canonical"));
+    }
+    let persisted: PersistedTierDeleteJournalEntry =
+        serde_json::from_slice(data).map_err(|err| Error::other(format!("decode tier delete journal failed: {err}")))?;
+    persisted.validate_legacy_recovery_shape()?;
+    let entry = persisted.into_jentry()?;
+    let Some((decoded_schema, _)) = legacy_tier_delete_recovery_descriptor(&entry) else {
+        return Err(Error::other("tier delete journal is not an exportable legacy version"));
+    };
+    if decoded_schema != source_schema || tier_delete_journal_object_name(&entry) != object_name {
+        return Err(Error::other("legacy tier delete journal identity does not match its recovery source"));
+    }
+    Ok(())
+}
+
 fn legacy_tier_delete_control_matches(
     control: &IlmRecoveryControl,
     identity: &IlmRecoveryControlIdentity,
@@ -6140,17 +6174,18 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        TIER_DELETE_DISPATCH_MANIFEST_VERSION, TIER_DELETE_DISPATCH_PARENT_RECORD_TYPE, TIER_DELETE_DISPATCH_PARENT_VERSION,
-        TIER_DELETE_JOURNAL_EXACT_VERSION, TIER_DELETE_JOURNAL_LEGACY_PREFIX, TIER_DELETE_JOURNAL_SOLE_OWNER_VERSION,
-        TIER_DELETE_JOURNAL_STATE_VERSION, TIER_DELETE_JOURNAL_TRANSACTION_VERSION, TIER_DELETE_JOURNAL_V6_PREFIX,
-        TierDeleteDispatchChunkBinding, TierDeleteDispatchManifest, TierDeleteDispatchManifestState, TierDeleteDispatchParent,
-        TierDeleteDispatchParentState, TierDeleteDispatchRecord, await_tier_delete_journal_recovery,
+        PersistedTierDeleteJournalEntry, TIER_DELETE_DISPATCH_MANIFEST_VERSION, TIER_DELETE_DISPATCH_PARENT_RECORD_TYPE,
+        TIER_DELETE_DISPATCH_PARENT_VERSION, TIER_DELETE_JOURNAL_EXACT_VERSION, TIER_DELETE_JOURNAL_LEGACY_PREFIX,
+        TIER_DELETE_JOURNAL_SOLE_OWNER_VERSION, TIER_DELETE_JOURNAL_STATE_VERSION, TIER_DELETE_JOURNAL_TRANSACTION_VERSION,
+        TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V6_PREFIX,
+        TIER_DELETE_JOURNAL_VERSION, TierDeleteDispatchChunkBinding, TierDeleteDispatchManifest, TierDeleteDispatchManifestState,
+        TierDeleteDispatchParent, TierDeleteDispatchParentState, TierDeleteDispatchRecord, await_tier_delete_journal_recovery,
         decode_tier_delete_dispatch_record, decode_tier_delete_journal_entry, encode_tier_delete_dispatch_manifest,
         encode_tier_delete_dispatch_parent, encode_tier_delete_journal_entry, object_info_references_tier_delete,
         record_tier_delete_journal_backend_identity, same_tier_delete_authorization_identity, same_tier_delete_journal_identity,
         tier_delete_dispatch_child_matches_parent, tier_delete_dispatch_chunk_manifest_object_name,
         tier_delete_dispatch_journal_set_digest, tier_delete_dispatch_manifest_object_name, tier_delete_journal_object_name,
-        tier_delete_source_matches_dispatch_scope,
+        tier_delete_source_matches_dispatch_scope, validate_legacy_tier_delete_recovery_source,
     };
     use crate::bucket::lifecycle::tier_sweeper::{
         Jentry, TierDeleteDispatchBinding, TierDeleteJournalState, TierDeleteSourceIdentity,
@@ -6609,6 +6644,72 @@ mod tests {
         }
     }
 
+    #[test]
+    fn legacy_recovery_export_rejects_fields_from_later_journal_versions() {
+        let later = bound_v6_journal_entry(TierDeleteJournalState::Prepared);
+        let v1 = PersistedTierDeleteJournalEntry {
+            version: 1,
+            obj_name: "remote/object".to_string(),
+            version_id: "opaque".to_string(),
+            tier_name: "WARM".to_string(),
+            backend_identity: None,
+            version_id_exact: None,
+            version_state: None,
+            state: None,
+            source: None,
+            dispatch: None,
+        };
+        let mut v2 = v1.clone();
+        v2.version = TIER_DELETE_JOURNAL_VERSION;
+        v2.backend_identity = Some([7; 32]);
+
+        let assert_rejected = |persisted: PersistedTierDeleteJournalEntry, schema: &str| {
+            let normalized = persisted
+                .clone()
+                .into_jentry()
+                .expect("the generic compatibility decoder should demonstrate the discarded field");
+            let object_name = tier_delete_journal_object_name(&normalized);
+            let encoded = serde_json::to_vec(&persisted).expect("mixed-version journal fixture should encode");
+            let err = validate_legacy_tier_delete_recovery_source(&object_name, schema, &encoded)
+                .expect_err("legacy recovery export must reject fields from later versions");
+            assert!(err.to_string().contains("later version"));
+        };
+
+        let mut invalid_v1 = Vec::new();
+        let mut with_backend = v1.clone();
+        with_backend.backend_identity = Some([7; 32]);
+        invalid_v1.push(with_backend);
+        for persisted in [&v1, &v2] {
+            let schema = if persisted.version == 1 {
+                TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA
+            } else {
+                TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA
+            };
+            let mut invalid = Vec::new();
+            let mut with_exact = persisted.clone();
+            with_exact.version_id_exact = Some(false);
+            invalid.push(with_exact);
+            let mut with_version_state = persisted.clone();
+            with_version_state.version_state = Some(rustfs_filemeta::TransitionVersionState::Unknown);
+            invalid.push(with_version_state);
+            let mut with_state = persisted.clone();
+            with_state.state = Some(TierDeleteJournalState::Committed);
+            invalid.push(with_state);
+            let mut with_source = persisted.clone();
+            with_source.source = later.source.clone();
+            invalid.push(with_source);
+            let mut with_dispatch = persisted.clone();
+            with_dispatch.dispatch = later.dispatch.clone();
+            invalid.push(with_dispatch);
+            for record in invalid {
+                assert_rejected(record, schema);
+            }
+        }
+        for record in invalid_v1 {
+            assert_rejected(record, TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA);
+        }
+    }
+
     #[test]
     fn tier_delete_journal_path_is_stable_and_sanitized() {
         let je = journal_entry();

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -1688,6 +1688,15 @@ impl PeerRestClient {
         Ok((self.topology_member.clone(), supported_version, epoch))
     }
 
+    pub async fn probe_ilm_recovery_export(&self, topology_fingerprint: String) -> Result<(String, Uuid)> {
+        let probe = rustfs_protos::ilm_recovery_export_capability_probe(Uuid::new_v4().as_bytes());
+        let result = self
+            .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint, probe)
+            .await?;
+        let epoch = decode_remote_version_state_capability(&self.topology_member, &result)?;
+        Ok((self.topology_member.clone(), epoch))
+    }
+
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
         let result = tokio::time::timeout(BUCKET_METADATA_RELOAD_TIMEOUT, async {
             let result = self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;

crates/ecstore/src/config/com.rs

@@ -449,14 +449,6 @@ where
     Ok(data)
 }
 
-pub(crate) async fn read_config_limited<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
-where
-    S: EcstoreObjectIO,
-{
-    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), false, Some(max_bytes)).await?;
-    Ok(data)
-}
-
 pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
 where
     S: EcstoreObjectIO,
@@ -465,6 +457,14 @@ where
     Ok(data)
 }
 
+pub(crate) async fn read_config_limited<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
+where
+    S: EcstoreObjectIO,
+{
+    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), false, Some(max_bytes)).await?;
+    Ok(data)
+}
+
 pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
     api: Arc<S>,
     file: &str,
@@ -476,6 +476,18 @@ where
     read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
 }
 
+pub(crate) async fn read_config_limited_preserve_empty_with_metadata_opts<S>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+    max_bytes: usize,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    read_config_with_metadata_inner(api, file, opts, true, Some(max_bytes)).await
+}
+
 /// Read an existing config object without treating an empty payload as absent.
 /// Callers that validate their own payload format need to distinguish corruption
 /// from `ConfigNotFound`.

crates/ecstore/src/services/notification_sys.rs

@@ -33,11 +33,11 @@ use rustfs_madmin::net::NetInfo;
 use rustfs_madmin::{ItemState, ServerProperties, StorageInfo};
 use rustfs_utils::XHost;
 use sha2::{Digest, Sha256};
-use std::collections::{BTreeMap, HashMap, hash_map::DefaultHasher};
+use std::collections::{BTreeMap, BTreeSet, HashMap, hash_map::DefaultHasher};
 use std::future::Future;
 use std::hash::{Hash, Hasher};
 use std::sync::{
-    Arc, Mutex, OnceLock,
+    Arc, LazyLock, Mutex, OnceLock,
     atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 use std::time::{Duration, Instant, SystemTime};
@@ -311,12 +311,20 @@ pub struct LegacyTransitionStateReconcileFleetProofToken {
     _permit: FleetCapabilityProofPermit,
 }
 
+/// Effect-window authority for one immutable ILM recovery export.
+pub struct IlmRecoveryExportFleetProofToken {
+    token: FleetCapabilityProofToken,
+    _permit: FleetCapabilityProofPermit,
+}
+
 static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static ILM_RECOVERY_EXPORT_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
+static ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
     CROSS_POOL_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
@@ -338,6 +346,10 @@ fn legacy_transition_state_reconcile_fleet_proof_slot() -> &'static std::sync::R
     LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
+fn ilm_recovery_export_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
+    ILM_RECOVERY_EXPORT_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+}
+
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -573,6 +585,117 @@ pub async fn legacy_transition_state_reconcile_fleet_proof_matches(
     .await
 }
 
+pub async fn acquire_ilm_recovery_export_fleet_proof() -> Option<IlmRecoveryExportFleetProofToken> {
+    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
+    let proof = {
+        let state = ilm_recovery_export_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        acquire_ilm_recovery_export_fleet_proof_from(&state, expected_topology, Instant::now())?
+    };
+    let observed = observe_ilm_recovery_export_fleet(expected_topology).await?;
+    let state = ilm_recovery_export_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    ilm_recovery_export_fleet_proof_matches_observation_at(&state, &proof, expected_topology, &observed, Instant::now())
+        .then_some(proof)
+}
+
+fn acquire_ilm_recovery_export_fleet_proof_from(
+    state: &FleetCapabilityProofState,
+    expected_topology: &str,
+    now: Instant,
+) -> Option<IlmRecoveryExportFleetProofToken> {
+    let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
+    let permit = state.proof.as_ref()?.generation.try_acquire()?;
+    Some(IlmRecoveryExportFleetProofToken { token, _permit: permit })
+}
+
+pub async fn ilm_recovery_export_fleet_proof_matches(proof: &IlmRecoveryExportFleetProofToken) -> bool {
+    let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
+        return false;
+    };
+    {
+        let state = ilm_recovery_export_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if !ilm_recovery_export_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
+            return false;
+        }
+    }
+    let Some(observed) = observe_ilm_recovery_export_fleet(expected_topology).await else {
+        return false;
+    };
+    let state = ilm_recovery_export_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    ilm_recovery_export_fleet_proof_matches_observation_at(&state, proof, expected_topology, &observed, Instant::now())
+}
+
+pub fn ilm_recovery_export_topology_generation(proof: &IlmRecoveryExportFleetProofToken) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(b"rustfs-ilm-recovery-export-topology-v1\0");
+    hasher.update(proof.token.topology_fingerprint.as_bytes());
+    rustfs_utils::crypto::hex(hasher.finalize().as_slice())
+}
+
+pub fn ilm_recovery_export_member_epochs_sha256(proof: &IlmRecoveryExportFleetProofToken) -> String {
+    let encoded = serde_json::to_vec(proof.token.peer_epochs.as_ref()).expect("member epoch map is JSON encodable");
+    let mut hasher = Sha256::new();
+    hasher.update(b"rustfs-ilm-recovery-export-members-v1\0");
+    hasher.update(encoded);
+    rustfs_utils::crypto::hex(hasher.finalize().as_slice())
+}
+
+pub fn ilm_recovery_export_local_process_epoch() -> Uuid {
+    *ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH
+}
+
+fn ilm_recovery_export_fleet_proof_matches_at(
+    state: &FleetCapabilityProofState,
+    proof: &IlmRecoveryExportFleetProofToken,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
+    proof._permit.generation.is_accepting()
+        && fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
+        && state
+            .proof
+            .as_ref()
+            .is_some_and(|current| Arc::ptr_eq(&current.generation, &proof._permit.generation))
+}
+
+fn ilm_recovery_export_fleet_proof_matches_observation_at(
+    state: &FleetCapabilityProofState,
+    proof: &IlmRecoveryExportFleetProofToken,
+    expected_topology: &str,
+    observed: &BTreeMap<String, Uuid>,
+    now: Instant,
+) -> bool {
+    ilm_recovery_export_fleet_proof_matches_at(state, proof, expected_topology, now)
+        && proof.token.peer_epochs.as_ref() == observed
+}
+
+async fn observe_ilm_recovery_export_fleet(expected_topology: &str) -> Option<BTreeMap<String, Uuid>> {
+    #[cfg(test)]
+    {
+        let state = ilm_recovery_export_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if fleet_capability_proof_valid_at(state.proof.as_ref(), expected_topology, Instant::now()) {
+            return state.proof.as_ref().map(|proof| proof.peer_epochs.as_ref().clone());
+        }
+    }
+    let notification_sys = get_global_notification_sys()?;
+    timeout(
+        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+        notification_sys.probe_ilm_recovery_export_fleet(expected_topology),
+    )
+    .await
+    .ok()?
+    .ok()
+}
+
 async fn legacy_transition_state_reconcile_fleet_proof_matches_with_observer<F, Fut>(
     slot: &std::sync::RwLock<FleetCapabilityProofState>,
     proof: &LegacyTransitionStateReconcileFleetProofToken,
@@ -661,7 +784,7 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() {
         state.proof.clone()
     } else {
         Some(FleetCapabilityProof::new(
-            topology,
+            topology.clone(),
             Arc::new(BTreeMap::new()),
             now + Duration::from_secs(60 * 60),
         ))
@@ -694,6 +817,21 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() {
     decommission_state.topology_conflict = false;
     decommission_state.draining_generation = None;
     decommission_state.proof = proof.as_ref().map(FleetCapabilityProof::with_fresh_generation);
+    drop(decommission_state);
+    let mut export_state = ilm_recovery_export_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    if !fleet_capability_proof_valid_at(export_state.proof.as_ref(), &topology, now) {
+        debug_assert!(
+            export_state
+                .proof
+                .as_ref()
+                .is_none_or(|current| current.generation.is_drained())
+        );
+        export_state.topology_conflict = false;
+        export_state.draining_generation = None;
+        export_state.proof = proof.as_ref().map(FleetCapabilityProof::with_fresh_generation);
+    }
 }
 
 #[cfg(test)]
@@ -950,6 +1088,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 tier_delete_journal_fleet_proof_slot(),
                 decommission_target_fence_fleet_proof_slot(),
                 legacy_transition_state_reconcile_fleet_proof_slot(),
+                ilm_recovery_export_fleet_proof_slot(),
             ] {
                 mark_fleet_capability_topology_conflict(slot);
             }
@@ -959,29 +1098,42 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
 
     tokio::spawn(async move {
         loop {
-            let result = match get_global_notification_sys() {
-                Some(notification_sys) => {
-                    match timeout(
+            let notification_sys = get_global_notification_sys();
+            let remote_version_state_probe = async {
+                match notification_sys.as_ref() {
+                    Some(notification_sys) => timeout(
                         REMOTE_VERSION_STATE_PROBE_TIMEOUT,
                         notification_sys.probe_remote_version_state_fleet(&topology_fingerprint),
                     )
                     .await
-                    {
-                        Ok(result) => result,
-                        Err(_) => Err(Error::other("remote version state fleet capability probe timed out")),
-                    }
+                    .unwrap_or_else(|_| Err(Error::other("remote version state fleet capability probe timed out"))),
+                    None => Err(Error::other("remote version state fleet capability notification system is unavailable")),
                 }
-                None => Err(Error::other("remote version state fleet capability notification system is unavailable")),
             };
-            let fence_probe = match get_global_notification_sys() {
-                Some(notification_sys) => timeout(
-                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                    notification_sys.probe_cross_pool_fence_fleet(&topology_fingerprint),
-                )
-                .await
-                .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
-                None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
+            let cross_pool_fence_probe = async {
+                match notification_sys.as_ref() {
+                    Some(notification_sys) => timeout(
+                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                        notification_sys.probe_cross_pool_fence_fleet(&topology_fingerprint),
+                    )
+                    .await
+                    .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
+                    None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
+                }
             };
+            let recovery_export_probe = async {
+                match notification_sys.as_ref() {
+                    Some(notification_sys) => timeout(
+                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                        notification_sys.probe_ilm_recovery_export_fleet(&topology_fingerprint),
+                    )
+                    .await
+                    .unwrap_or_else(|_| Err(Error::other("ILM recovery export fleet capability probe timed out"))),
+                    None => Err(Error::other("ILM recovery export fleet capability notification system is unavailable")),
+                }
+            };
+            let (result, fence_probe, recovery_export_result) =
+                tokio::join!(remote_version_state_probe, cross_pool_fence_probe, recovery_export_probe);
             let (fence_result, journal_result, decommission_target_fence_result, reconcile_result) = match fence_probe {
                 Ok((peer_epochs, minimum_version)) => cross_pool_fence_policy_results(peer_epochs, minimum_version),
                 Err(err) => {
@@ -1004,6 +1156,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
                 revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
                 revoke_fleet_capability_proof(legacy_transition_state_reconcile_fleet_proof_slot());
+                revoke_fleet_capability_proof(ilm_recovery_export_fleet_proof_slot());
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -1030,6 +1183,24 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                     "notification capability probe"
                 );
             }
+            if !topology_conflict
+                && let Some(err) = publish_fleet_capability_probe_result(
+                    ilm_recovery_export_fleet_proof_slot(),
+                    &topology_fingerprint,
+                    recovery_export_result,
+                    Instant::now(),
+                )
+            {
+                debug!(
+                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                    capability = "ilm_recovery_export_v1",
+                    state = "failed_closed",
+                    error = %err,
+                    "notification capability probe"
+                );
+            }
             if !topology_conflict
                 && let Some(err) = publish_fleet_capability_probe_result(
                     tier_delete_journal_fleet_proof_slot(),
@@ -1174,6 +1345,46 @@ impl NotificationSys {
         }
         Ok((peer_epochs, minimum_version))
     }
+
+    async fn probe_ilm_recovery_export_fleet(&self, topology_fingerprint: &str) -> Result<BTreeMap<String, Uuid>> {
+        if self.peer_clients.len() != self.peer_topology_hosts.len() {
+            return Err(Error::other("ILM recovery export capability fleet membership is incomplete"));
+        }
+        let local_member = runtime_sources::local_node_name().await;
+        if local_member.trim().is_empty() {
+            return Err(Error::other("ILM recovery export local member identity is unavailable"));
+        }
+        let mut peer_epochs = BTreeMap::new();
+        insert_remote_version_state_peer(&mut peer_epochs, local_member.clone(), ilm_recovery_export_local_process_epoch())?;
+        let probes = self.peer_clients.iter().map(|client| async {
+            let client = client
+                .as_ref()
+                .ok_or_else(|| Error::other("ILM recovery export capability peer is unreachable"))?;
+            client.probe_ilm_recovery_export(topology_fingerprint.to_string()).await
+        });
+        for result in join_all(probes).await {
+            let (peer, epoch) = result?;
+            insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
+        }
+        validate_ilm_recovery_export_members(&self.peer_topology_hosts, &local_member, &peer_epochs)?;
+        Ok(peer_epochs)
+    }
+}
+
+fn validate_ilm_recovery_export_members(
+    expected_remote_members: &[String],
+    local_member: &str,
+    observed: &BTreeMap<String, Uuid>,
+) -> Result<()> {
+    let expected = expected_remote_members
+        .iter()
+        .cloned()
+        .chain(std::iter::once(local_member.to_string()))
+        .collect::<BTreeSet<_>>();
+    if expected.len() != expected_remote_members.len().saturating_add(1) || observed.keys().ne(expected.iter()) {
+        return Err(Error::other("ILM recovery export capability fleet membership does not match topology"));
+    }
+    Ok(())
 }
 
 /// Rolling tier activity summed over every cluster member that answered, with
@@ -3507,6 +3718,81 @@ mod tests {
         assert!(captured != restarted.token());
     }
 
+    #[test]
+    fn ilm_recovery_export_member_digest_is_order_independent_and_epoch_bound() {
+        let now = Instant::now();
+        let local_epoch = ilm_recovery_export_local_process_epoch();
+        assert!(!local_epoch.is_nil());
+        assert_eq!(local_epoch, ilm_recovery_export_local_process_epoch());
+        let remote_epoch = Uuid::new_v4();
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let peers = BTreeMap::from([("node-b".to_string(), remote_epoch), ("node-a".to_string(), local_epoch)]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers), now).is_none());
+        let proof = {
+            let state = slot.read().expect("export proof slot should not poison");
+            acquire_ilm_recovery_export_fleet_proof_from(&state, "topology-a", now).expect("complete fleet should admit export")
+        };
+        let digest = ilm_recovery_export_member_epochs_sha256(&proof);
+
+        let changed_slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let changed = BTreeMap::from([("node-a".to_string(), local_epoch), ("node-b".to_string(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&changed_slot, "topology-a", Ok(changed), now).is_none());
+        let changed_proof = {
+            let state = changed_slot.read().expect("export proof slot should not poison");
+            acquire_ilm_recovery_export_fleet_proof_from(&state, "topology-a", now).expect("complete fleet should admit export")
+        };
+        assert_ne!(digest, ilm_recovery_export_member_epochs_sha256(&changed_proof));
+    }
+
+    #[test]
+    fn ilm_recovery_export_members_must_match_the_exact_topology() {
+        let expected_remote = vec!["node-b".to_string()];
+        let local = "node-a";
+        let complete = BTreeMap::from([
+            (local.to_string(), Uuid::new_v4()),
+            (expected_remote[0].clone(), Uuid::new_v4()),
+        ]);
+        assert!(validate_ilm_recovery_export_members(&expected_remote, local, &complete).is_ok());
+
+        let unexpected = BTreeMap::from([(local.to_string(), Uuid::new_v4()), ("node-c".to_string(), Uuid::new_v4())]);
+        assert!(validate_ilm_recovery_export_members(&expected_remote, local, &unexpected).is_err());
+        assert!(
+            validate_ilm_recovery_export_members(&[local.to_string()], local, &complete).is_err(),
+            "the configured remote set cannot repeat the local member"
+        );
+    }
+
+    #[test]
+    fn ilm_recovery_export_restart_revokes_authority_until_permit_drains() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let original = BTreeMap::from([("node-a".to_string(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(original), now).is_none());
+        let admitted = {
+            let state = slot.read().expect("export proof slot should not poison");
+            acquire_ilm_recovery_export_fleet_proof_from(&state, "topology-a", now).expect("fresh fleet should admit export")
+        };
+
+        let restarted = BTreeMap::from([("node-a".to_string(), Uuid::new_v4())]);
+        let draining = publish_fleet_capability_probe_result(&slot, "topology-a", Ok(restarted.clone()), now)
+            .expect("restart must wait for the admitted export effect window");
+        assert!(draining.to_string().contains("previous generation to drain"));
+        {
+            let state = slot.read().expect("export proof slot should not poison");
+            assert!(!ilm_recovery_export_fleet_proof_matches_at(&state, &admitted, "topology-a", now));
+            assert!(
+                acquire_ilm_recovery_export_fleet_proof_from(&state, "topology-a", now).is_none(),
+                "successor authority must wait for the old effect window to drain"
+            );
+        }
+        drop(admitted);
+        assert!(
+            publish_fleet_capability_probe_result(&slot, "topology-a", Ok(restarted), now + Duration::from_millis(1)).is_none()
+        );
+        let state = slot.read().expect("export proof slot should not poison");
+        assert!(acquire_ilm_recovery_export_fleet_proof_from(&state, "topology-a", now).is_some());
+    }
+
     #[test]
     fn tier_delete_journal_generation_is_stable_across_members_and_process_restarts() {
         let topology = "topology-a";

crates/ecstore/src/store/init.rs

@@ -830,6 +830,10 @@ mod tests {
                 IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, list_recovery_controls, load_recovery_control,
                 observe_recovery_source, save_recovery_control_if_absent,
             },
+            recovery_export::{
+                create_recovery_export, inspect_recovery_export_observation, load_recovery_export,
+                recovery_export_record_object_name,
+            },
             tier_delete_journal::{
                 DecommissionCheckpointTargetFailureHook, TIER_DELETE_DISPATCH_MANIFEST_PREFIX, TIER_DELETE_JOURNAL_PREFIX,
                 TierDeleteChunkTestBarrier, TierDeleteChunkTestStage, TierDeleteDispatchBatchLimitGuard,
@@ -16870,6 +16874,43 @@ mod tests {
             }
         }
 
+        let creator_sha256 = rustfs_utils::crypto::hex_sha256(b"legacy-export-actor", ToOwned::to_owned);
+        let mut created_exports = Vec::new();
+        for exportable in first_controls
+            .iter()
+            .filter(|control| control.classification == IlmRecoveryClassification::RetainedAmbiguous)
+        {
+            let observation = inspect_recovery_export_observation(store.clone(), &exportable.control_id)
+                .await
+                .expect("fresh legacy recovery observation should be exportable");
+            let created = create_recovery_export(store.clone(), &observation, &creator_sha256)
+                .await
+                .expect("legacy recovery export should be created exactly once");
+            assert!(!created.replayed);
+            let loaded = load_recovery_export(store.clone(), &created.export_id)
+                .await
+                .expect("created legacy recovery export should load");
+            assert_eq!(loaded.encoded, created.encoded, "export readback must preserve the exact committed bytes");
+            let replayed = create_recovery_export(store.clone(), &observation, &creator_sha256)
+                .await
+                .expect("the same observed generation should replay its immutable export");
+            assert!(replayed.replayed);
+            assert_eq!(replayed.encoded, created.encoded);
+            created_exports.push(created);
+        }
+        assert_eq!(created_exports.len(), 2, "both v1 and v2 legacy journals must have an export path");
+
+        let corrupt_export_id = &created_exports[0].export_id;
+        let export_path = recovery_export_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, corrupt_export_id)
+            .expect("export path should build");
+        com::save_config(store.clone(), &export_path, Vec::new())
+            .await
+            .expect("zero-byte corruption fixture should persist");
+        let corrupt_export = load_recovery_export(store.clone(), corrupt_export_id)
+            .await
+            .expect_err("an existing zero-byte export must fail closed");
+        assert!(!matches!(corrupt_export, Error::ConfigNotFound));
+
         com::save_config(
             store.clone(),
             &journal_paths[0],