From 7c0e1f2ef0482132ab3e8b08e1c8a63e9696f638 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 6 Sep 2026 15:48:02 +0800 Subject: [PATCH] feat(ilm): add immutable legacy recovery exports --- crates/ecstore/src/api/mod.rs | 14 +- .../src/bucket/lifecycle/config_boundary.rs | 35 + .../src/bucket/lifecycle/durable_namespace.rs | 44 +- crates/ecstore/src/bucket/lifecycle/mod.rs | 1 + .../src/bucket/lifecycle/recovery_control.rs | 2 +- .../src/bucket/lifecycle/recovery_export.rs | 837 ++++++++++++++++++ .../bucket/lifecycle/tier_delete_journal.rs | 117 ++- .../src/cluster/rpc/peer_rest_client.rs | 9 + crates/ecstore/src/config/com.rs | 28 +- .../ecstore/src/services/notification_sys.rs | 324 ++++++- crates/ecstore/src/store/init.rs | 41 + crates/protos/src/lib.rs | 39 +- rustfs/src/admin/handlers/ilm_transition.rs | 467 +++++++++- rustfs/src/admin/route_policy.rs | 14 + rustfs/src/admin/route_registration_test.rs | 20 + rustfs/src/admin/storage_api.rs | 5 +- rustfs/src/storage/rpc/node_service.rs | 52 +- rustfs/src/storage/storage_api.rs | 8 +- 18 files changed, 1982 insertions(+), 75 deletions(-) create mode 100644 crates/ecstore/src/bucket/lifecycle/recovery_export.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index e0cc06d26..0fbe6137a 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, }; diff --git a/crates/ecstore/src/bucket/lifecycle/config_boundary.rs b/crates/ecstore/src/bucket/lifecycle/config_boundary.rs index dc0fa8a83..12b2e2ba5 100644 --- a/crates/ecstore/src/bucket/lifecycle/config_boundary.rs +++ b/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(api: Arc, file: &str, max_bytes: usize) -> Result> +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(api: Arc, file: &str, opts: &ObjectOptions) -> Result<(Vec, 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( + api: Arc, + file: &str, + opts: &ObjectOptions, + max_bytes: usize, +) -> Result<(Vec, 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(api: Arc, file: &str, data: Vec) -> Result<()> where S: ObjectIO< diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 58e7ae607..2a785247c 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/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, }, + 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 { + 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()))?; diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index 3eb93cca4..2d63efb5b 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/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; diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_control.rs b/crates/ecstore/src/bucket/lifecycle/recovery_control.rs index 599849ca9..de6224e82 100644 --- a/crates/ecstore/src/bucket/lifecycle/recovery_control.rs +++ b/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")); } diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_export.rs b/crates/ecstore/src/bucket/lifecycle/recovery_export.rs new file mode 100644 index 000000000..52048f476 --- /dev/null +++ b/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, + 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> { + 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 { + 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 { + 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, control_id: &str) -> Result { + 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, + observation: &IlmRecoveryExportObservation, + creator_sha256: &str, +) -> Result { + 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 ¤t != 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(¤t.control_id, ¤t.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(¤t, creator_sha256, created_at_unix_nanos, ¤t_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, export_id: &str) -> Result { + let (created, _) = load_recovery_export_decoded(api, export_id).await?; + Ok(created) +} + +async fn load_recovery_export_decoded( + api: Arc, + 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, control_id: &str) -> Result { + load_exportable_control_with_options(api, control_id, &ObjectOptions::default()).await +} + +async fn load_exportable_control_no_lock(api: Arc, control_id: &str) -> Result { + load_exportable_control_with_options( + api, + control_id, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await +} + +async fn load_exportable_control_with_options( + api: Arc, + control_id: &str, + options: &ObjectOptions, +) -> Result { + 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, + expected: &IlmRecoveryExportObservation, + proof: &crate::services::notification_sys::IlmRecoveryExportFleetProofToken, +) -> Result<(IlmRecoveryExportObservation, Vec)> { + 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, + canonical_path: &str, + source_schema: &str, +) -> Result { + 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, + canonical_path: &str, + source_schema: &str, +) -> Result { + let mut copies = Vec::new(); + let mut canonical: Option<(String, String, Vec)> = 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 { + 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 { + 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) -> Result { + 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, next: String) -> Result { + 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::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 { + 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()); + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 03b04a8a2..18613523b 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/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 { 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(); diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index c6a15a6b9..f6f5f58e8 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/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; diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 8c5132487..4081a3e3a 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -449,14 +449,6 @@ where Ok(data) } -pub(crate) async fn read_config_limited(api: Arc, file: &str, max_bytes: usize) -> Result> -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(api: Arc, file: &str, max_bytes: usize) -> Result> where S: EcstoreObjectIO, @@ -465,6 +457,14 @@ where Ok(data) } +pub(crate) async fn read_config_limited(api: Arc, file: &str, max_bytes: usize) -> Result> +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( api: Arc, 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( + api: Arc, + file: &str, + opts: &ObjectOptions, + max_bytes: usize, +) -> Result<(Vec, 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`. diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index d1955a397..7769ea1b1 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/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> = OnceLock::new(); static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock> = OnceLock::new(); static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock> = OnceLock::new(); static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock> = OnceLock::new(); static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock> = OnceLock::new(); +static ILM_RECOVERY_EXPORT_FLEET_PROOF: OnceLock> = OnceLock::new(); static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock = OnceLock::new(); +static ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4); fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock { 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 { + 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 { + 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 { + 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(¤t.generation, &proof._permit.generation)) +} + +fn ilm_recovery_export_fleet_proof_matches_observation_at( + state: &FleetCapabilityProofState, + proof: &IlmRecoveryExportFleetProofToken, + expected_topology: &str, + observed: &BTreeMap, + 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> { + #[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( slot: &std::sync::RwLock, 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> { + 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, +) -> Result<()> { + let expected = expected_remote_members + .iter() + .cloned() + .chain(std::iter::once(local_member.to_string())) + .collect::>(); + 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"; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 86662f182..6ec5c3c00 100644 --- a/crates/ecstore/src/store/init.rs +++ b/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], diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 81ead7d65..47b7811e1 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -175,6 +175,7 @@ pub const BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION: u32 = 2; pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v3\0"; pub const REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-tier-remote-version-state-capability-v1\0"; pub const CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-cross-pool-fence-capability-v1\0"; +pub const ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-ilm-recovery-export-capability-v1\0"; pub const TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE: usize = 64 * 1024; pub const TIER_MUTATION_RPC_MAX_COMMIT_PAYLOAD_SIZE: usize = 1024; pub const TIER_MUTATION_RPC_MAX_ABORT_PAYLOAD_SIZE: usize = TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE; @@ -219,6 +220,18 @@ pub fn is_cross_pool_fence_capability_probe(command: &[u8]) -> bool { && command.starts_with(CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX) } +pub fn ilm_recovery_export_capability_probe(nonce: &[u8; 16]) -> Vec { + let mut probe = Vec::with_capacity(ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX.len() + nonce.len()); + probe.extend_from_slice(ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX); + probe.extend_from_slice(nonce); + probe +} + +pub fn is_ilm_recovery_export_capability_probe(command: &[u8]) -> bool { + command.len() == ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX.len() + 16 + && command.starts_with(ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX) +} + pub fn encode_remote_version_state_capability( topology_member: &str, process_epoch: &[u8; 16], @@ -2127,12 +2140,13 @@ mod scanner_activity_tests { mod heal_control_tests { use super::{ CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, - REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX, canonical_heal_control_capability_ack, canonical_heal_control_request_body, - canonical_heal_control_response_body, decode_remote_version_state_capability, encode_cross_pool_fence_capability, - encode_remote_version_state_capability, heal_control_capability_probe, heal_control_coordinator_epoch, - heal_control_execution_timeout, heal_control_execution_timeout_for, internode_rpc_timeout, - is_cross_pool_fence_capability_probe, is_heal_control_capability_probe, is_remote_version_state_capability_probe, - normalize_internode_rpc_timeout, remote_version_state_capability_probe, + ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX, REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX, + canonical_heal_control_capability_ack, canonical_heal_control_request_body, canonical_heal_control_response_body, + decode_remote_version_state_capability, encode_cross_pool_fence_capability, encode_remote_version_state_capability, + heal_control_capability_probe, heal_control_coordinator_epoch, heal_control_execution_timeout, + heal_control_execution_timeout_for, ilm_recovery_export_capability_probe, internode_rpc_timeout, + is_cross_pool_fence_capability_probe, is_heal_control_capability_probe, is_ilm_recovery_export_capability_probe, + is_remote_version_state_capability_probe, normalize_internode_rpc_timeout, remote_version_state_capability_probe, }; use crate::heal_control; use std::time::Duration; @@ -2196,6 +2210,19 @@ mod heal_control_tests { assert!(!is_remote_version_state_capability_probe(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX)); } + #[test] + fn ilm_recovery_export_capability_probe_requires_exact_prefix_and_nonce() { + let probe = ilm_recovery_export_capability_probe(&[7; 16]); + assert!(is_ilm_recovery_export_capability_probe(&probe)); + assert!(!is_ilm_recovery_export_capability_probe(ILM_RECOVERY_EXPORT_CAPABILITY_PROBE_PREFIX)); + let mut wrong_prefix = probe.clone(); + wrong_prefix[0] ^= 1; + assert!(!is_ilm_recovery_export_capability_probe(&wrong_prefix)); + let mut extra = probe; + extra.push(0); + assert!(!is_ilm_recovery_export_capability_probe(&extra)); + } + #[test] fn remote_version_state_capability_binds_member_and_process_epoch() { let encoded = diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index 74788d0d9..3562fecc6 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -14,18 +14,19 @@ use crate::admin::auth::authorize_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::admin::runtime_sources::object_store_from_extensions; +use crate::admin::runtime_sources::{current_action_credentials, object_store_from_extensions}; use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; use crate::admin::storage_api::error::StorageError; use crate::admin::storage_api::lifecycle::{ - IlmRecoveryClassification, IlmRecoveryProtocol, ManualTransitionCancelCheck, ManualTransitionJobRecord, - ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot, ManualTransitionRunOptions, - ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, - TransitionOperatorDeleteResult, TransitionOperatorError, claim_manual_transition_scope_admission, - delete_manual_transition_scope_admission_if_current, delete_transition_candidate_for_operator, - enqueue_transition_for_existing_objects_scoped, finalize_missing_transition_transaction_for_operator, - inspect_recovery_control, inspect_transition_transaction_for_operator, list_recovery_controls, - load_manual_transition_job_record, load_manual_transition_scope_admission, manual_transition_job_lease_expired, + IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryExportObservation, IlmRecoveryProtocol, + ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink, + ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission, + ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError, + claim_manual_transition_scope_admission, create_recovery_export, delete_manual_transition_scope_admission_if_current, + delete_transition_candidate_for_operator, enqueue_transition_for_existing_objects_scoped, + finalize_missing_transition_transaction_for_operator, inspect_recovery_control, inspect_recovery_export_observation, + inspect_transition_transaction_for_operator, list_recovery_controls, load_manual_transition_job_record, + load_manual_transition_scope_admission, load_recovery_export, manual_transition_job_lease_expired, manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired, persist_manual_transition_job_progress_if_owned, renew_manual_transition_job_lease_if_owned, request_manual_transition_job_cancel, save_manual_transition_job_record, update_manual_transition_job_record, @@ -34,21 +35,30 @@ use crate::admin::storage_api::runtime::ECStore; use crate::admin::storage_api::s3::{S3ErrorCode as AdminS3ErrorCode, error as admin_s3_error}; use crate::admin::utils::json_response; use crate::server::{ADMIN_PREFIX, RemoteAddr}; -use http::HeaderMap; +use aes_gcm::{ + Aes256Gcm, Key, Nonce, + aead::{Aead, KeyInit}, +}; +use http::{HeaderMap, HeaderValue, header}; use hyper::{Method, StatusCode}; use matchit::Params; +use rand::RngExt; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_utils::{ - MaskedAccessKey, + MaskedAccessKey, base64_decode_url_safe_no_pad, base64_encode_url_safe_no_pad, + crypto::hex_sha256, http::{AMZ_REQUEST_ID, REQUEST_ID_HEADER}, }; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::pin::Pin; use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; use std::time::Duration as StdDuration; +use time::{Duration, OffsetDateTime}; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use uuid::Uuid; @@ -60,6 +70,8 @@ const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_ILM_TRANSITION: &str = "ilm_transition"; const EVENT_ADMIN_ILM_TRANSITION_STATE: &str = "admin_ilm_transition_state"; const EVENT_ADMIN_ILM_TRANSITION_RECONCILE: &str = "admin_ilm_transition_reconcile"; +const ILM_RECOVERY_OBSERVATION_RECEIPT_TTL: Duration = Duration::minutes(15); +const MAX_ILM_RECOVERY_RECEIPT_SIZE: usize = 32 * 1024; static ACTIVE_MANUAL_TRANSITION_SCOPES: OnceLock>> = OnceLock::new(); #[cfg(feature = "e2e-test-hooks")] @@ -242,6 +254,16 @@ pub fn register_ilm_transition_route(r: &mut S3Router) -> std::i format!("{ADMIN_PREFIX}/v3/ilm/recovery/records/{{control_id}}").as_str(), AdminOperation(&IlmRecoveryControlInspectHandler {}), )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/ilm/recovery/records/{{control_id}}").as_str(), + AdminOperation(&IlmRecoveryExportCreateHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/ilm/recovery/exports/{{export_id}}").as_str(), + AdminOperation(&IlmRecoveryExportDownloadHandler {}), + )?; Ok(()) } @@ -484,6 +506,203 @@ fn map_recovery_control_error(err: StorageError) -> S3Error { } } +fn recovery_export_id_from_params(params: &Params<'_, '_>) -> S3Result { + let export_id = params.get("export_id").unwrap_or(""); + if export_id.len() != 64 + || !export_id + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery export id")); + } + Ok(export_id.to_string()) +} + +fn map_recovery_export_error(err: StorageError) -> S3Error { + if err == StorageError::ConfigNotFound { + admin_s3_error(AdminS3ErrorCode::NoSuchKey, "ILM recovery export not found") + } else if err == StorageError::SlowDown { + admin_s3_error(AdminS3ErrorCode::SlowDown, "ILM recovery export admission capacity is exhausted") + } else if err == StorageError::PreconditionFailed { + admin_s3_error(AdminS3ErrorCode::OperationAborted, "ILM recovery export observation is stale") + } else { + admin_s3_error(AdminS3ErrorCode::OperationAborted, "ILM recovery export request cannot proceed") + } +} + +fn recovery_export_download_headers(export_id: &str, encoded_len: usize) -> S3Result { + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("application/json")); + headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + headers.insert( + header::CONTENT_DISPOSITION, + HeaderValue::from_str(&format!("attachment; filename=\"ilm-recovery-export-{export_id}.json\"")) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "invalid ILM recovery export filename"))?, + ); + headers.insert( + header::CONTENT_LENGTH, + HeaderValue::from_str(&encoded_len.to_string()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "invalid ILM recovery export length"))?, + ); + Ok(headers) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct IlmRecoveryObservationReceipt { + schema: String, + action: String, + actor_sha256: String, + issued_at_unix_nanos: i64, + expires_at_unix_nanos: i64, + nonce: Uuid, + observation: IlmRecoveryExportObservation, +} + +#[derive(Debug, Serialize)] +struct IlmRecoveryControlInspectResponse { + #[serde(flatten)] + control: IlmRecoveryControlView, + export_ready: bool, + #[serde(skip_serializing_if = "Option::is_none")] + export_not_ready_reason: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + observation_receipt: Option, + #[serde(skip_serializing_if = "Option::is_none")] + observation_receipt_expires_at_unix_nanos: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct IlmRecoveryExportCreateRequest { + action: String, + observation_receipt: String, +} + +#[derive(Debug, Serialize)] +struct IlmRecoveryExportCreateResponse { + export_id: String, + export_sha256: String, + download_url: String, + outcome: &'static str, +} + +fn recovery_actor_sha256(req: &S3Request) -> S3Result { + let access_key = &req + .credentials + .as_ref() + .ok_or_else(|| admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"))? + .access_key; + let mut bound = Vec::with_capacity(access_key.len() + 40); + bound.extend_from_slice(b"rustfs-ilm-recovery-actor-v1\0"); + bound.extend_from_slice(access_key.as_bytes()); + Ok(hex_sha256(&bound, ToOwned::to_owned)) +} + +fn recovery_receipt_credentials() -> S3Result { + current_action_credentials() + .filter(|credentials| !credentials.secret_key.is_empty()) + .ok_or_else(|| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt key is unavailable")) +} + +fn recovery_receipt_key(credentials: &Credentials) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(b"rustfs-ilm-recovery-observation-receipt-v1\0"); + hasher.update(credentials.secret_key.as_bytes()); + hasher.finalize().into() +} + +fn encode_recovery_receipt(payload: &IlmRecoveryObservationReceipt, credentials: &Credentials) -> S3Result { + let payload = serde_json::to_vec(payload) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "failed to encode ILM recovery receipt"))?; + if payload.len() > MAX_ILM_RECOVERY_RECEIPT_SIZE { + return Err(admin_s3_error( + AdminS3ErrorCode::InternalError, + "ILM recovery receipt exceeds maximum size", + )); + } + let cipher = Aes256Gcm::new(&Key::::from(recovery_receipt_key(credentials))); + let mut nonce_bytes = [0_u8; 12]; + rand::rng().fill(&mut nonce_bytes); + let ciphertext = cipher + .encrypt(&Nonce::from(nonce_bytes), payload.as_slice()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "failed to seal ILM recovery receipt"))?; + Ok(format!( + "{}.{}", + base64_encode_url_safe_no_pad(&nonce_bytes), + base64_encode_url_safe_no_pad(&ciphertext) + )) +} + +fn decode_recovery_receipt(token: &str, credentials: &Credentials) -> S3Result { + if token.len() > MAX_ILM_RECOVERY_RECEIPT_SIZE * 2 { + return Err(admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt")); + } + let Some((nonce, ciphertext)) = token.split_once('.') else { + return Err(admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt")); + }; + if ciphertext.contains('.') { + return Err(admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt")); + } + let nonce = base64_decode_url_safe_no_pad(nonce.as_bytes()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt"))?; + let nonce: [u8; 12] = nonce + .try_into() + .map_err(|_| admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt"))?; + let ciphertext = base64_decode_url_safe_no_pad(ciphertext.as_bytes()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt"))?; + let cipher = Aes256Gcm::new(&Key::::from(recovery_receipt_key(credentials))); + let plaintext = cipher + .decrypt(&Nonce::from(nonce), ciphertext.as_slice()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt"))?; + serde_json::from_slice(&plaintext) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt")) +} + +fn issue_recovery_observation_receipt( + observation: IlmRecoveryExportObservation, + actor_sha256: String, + now: OffsetDateTime, +) -> S3Result<(String, i64)> { + let expires_at = now + ILM_RECOVERY_OBSERVATION_RECEIPT_TTL; + let receipt = IlmRecoveryObservationReceipt { + schema: "rustfs-ilm-recovery-observation-receipt-v1".to_string(), + action: "export".to_string(), + actor_sha256, + issued_at_unix_nanos: i64::try_from(now.unix_timestamp_nanos()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt timestamp is invalid"))?, + expires_at_unix_nanos: i64::try_from(expires_at.unix_timestamp_nanos()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt timestamp is invalid"))?, + nonce: Uuid::new_v4(), + observation, + }; + let token = encode_recovery_receipt(&receipt, &recovery_receipt_credentials()?)?; + Ok((token, receipt.expires_at_unix_nanos)) +} + +fn validate_recovery_observation_receipt( + receipt: IlmRecoveryObservationReceipt, + actor_sha256: &str, + control_id: &str, + now_unix_nanos: i64, +) -> S3Result { + let ttl_nanos = i64::try_from(ILM_RECOVERY_OBSERVATION_RECEIPT_TTL.whole_nanoseconds()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt TTL is invalid"))?; + if receipt.schema != "rustfs-ilm-recovery-observation-receipt-v1" + || receipt.action != "export" + || receipt.actor_sha256 != actor_sha256 + || receipt.observation.control_id != control_id + || receipt.nonce.is_nil() + || receipt.issued_at_unix_nanos <= 0 + || receipt.issued_at_unix_nanos > now_unix_nanos + || receipt.expires_at_unix_nanos <= now_unix_nanos + || receipt.expires_at_unix_nanos.checked_sub(receipt.issued_at_unix_nanos) != Some(ttl_nanos) + { + return Err(admin_s3_error(AdminS3ErrorCode::AccessDenied, "invalid or expired ILM recovery receipt")); + } + Ok(receipt.observation) +} + fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error { match err { TransitionOperatorError::NotFound => s3_error!(NoSuchKey, "transition transaction not found"), @@ -1115,14 +1334,87 @@ pub struct IlmRecoveryControlInspectHandler {} impl Operation for IlmRecoveryControlInspectHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?; + let actor_sha256 = recovery_actor_sha256(&req)?; let control_id = recovery_control_id_from_params(¶ms)?; let Some(store) = object_store_from_extensions(&req.extensions) else { return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized")); }; - let control = inspect_recovery_control(store, &control_id) + let control = inspect_recovery_control(store.clone(), &control_id) .await .map_err(map_recovery_control_error)?; - json_response(StatusCode::OK, &control) + let now = OffsetDateTime::now_utc(); + let (export_ready, export_not_ready_reason, observation_receipt, expires_at) = + match inspect_recovery_export_observation(store, &control_id).await { + Ok(observation) => match issue_recovery_observation_receipt(observation, actor_sha256, now) { + Ok((token, expires_at)) => (true, None, Some(token), Some(expires_at)), + Err(_) => (false, Some("receipt_key_unavailable"), None, None), + }, + Err(_) => (false, Some("fleet_or_source_not_ready"), None, None), + }; + json_response( + StatusCode::OK, + &IlmRecoveryControlInspectResponse { + control, + export_ready, + export_not_ready_reason, + observation_receipt, + observation_receipt_expires_at_unix_nanos: expires_at, + }, + ) + } +} + +pub struct IlmRecoveryExportCreateHandler {} + +#[async_trait::async_trait] +impl Operation for IlmRecoveryExportCreateHandler { + async fn call(&self, mut req: S3Request, params: Params<'_, '_>) -> S3Result> { + authorize_transition_admin_request(&req, AdminAction::SetTierAction).await?; + let actor_sha256 = recovery_actor_sha256(&req)?; + let control_id = recovery_control_id_from_params(¶ms)?; + let Some(store) = object_store_from_extensions(&req.extensions) else { + return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized")); + }; + let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|_| { + admin_s3_error(AdminS3ErrorCode::InvalidRequest, "ILM recovery export body is too large or unreadable") + })?; + let request: IlmRecoveryExportCreateRequest = serde_json::from_slice(&body) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery export request"))?; + if request.action != "export" { + return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "unsupported ILM recovery action")); + } + let receipt = decode_recovery_receipt(&request.observation_receipt, &recovery_receipt_credentials()?)?; + let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt timestamp is invalid"))?; + let observation = validate_recovery_observation_receipt(receipt, &actor_sha256, &control_id, now)?; + let created = create_recovery_export(store, &observation, &actor_sha256) + .await + .map_err(map_recovery_export_error)?; + let response = IlmRecoveryExportCreateResponse { + download_url: format!("{ADMIN_PREFIX}/v3/ilm/recovery/exports/{}", created.export_id), + outcome: if created.replayed { "replayed" } else { "created" }, + export_id: created.export_id, + export_sha256: created.content_sha256, + }; + json_response(StatusCode::OK, &response) + } +} + +pub struct IlmRecoveryExportDownloadHandler {} + +#[async_trait::async_trait] +impl Operation for IlmRecoveryExportDownloadHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + authorize_transition_admin_request(&req, AdminAction::SetTierAction).await?; + let export_id = recovery_export_id_from_params(¶ms)?; + let Some(store) = object_store_from_extensions(&req.extensions) else { + return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized")); + }; + let export = load_recovery_export(store, &export_id) + .await + .map_err(map_recovery_export_error)?; + let headers = recovery_export_download_headers(&export_id, export.encoded.len())?; + Ok(S3Response::with_headers((StatusCode::OK, Body::from(export.encoded)), headers)) } } @@ -1242,6 +1534,155 @@ mod tests { }); } + #[test] + fn recovery_observation_receipt_is_opaque_actor_bound_and_tamper_evident() { + assert_eq!(ILM_RECOVERY_OBSERVATION_RECEIPT_TTL.whole_seconds(), 900); + let control_id = "ab".repeat(32); + let content_sha256 = hex_sha256(b"legacy", ToOwned::to_owned); + let copy_set_sha256 = hex_sha256( + &serde_json::to_vec(&serde_json::json!([{ + "authority": "pool-0/set-0", + "canonical_path": "ilm/tier-delete-journal/legacy.json", + "etag": "etag-a", + "encoded_len": 6, + "content_sha256": content_sha256, + }])) + .unwrap(), + ToOwned::to_owned, + ); + let observation: IlmRecoveryExportObservation = serde_json::from_value(serde_json::json!({ + "control_id": control_id, + "protocol": "tier_delete_journal", + "control_etag": "control-etag", + "control_revision": 1, + "classification": "retained_ambiguous", + "canonical_source_path": "ilm/tier-delete-journal/legacy.json", + "source_generation": { + "source_schema": "rustfs-tier-delete-journal-v1", + "source_etag": "etag-a", + "content_sha256": content_sha256, + "copy_set_sha256": copy_set_sha256, + "copies": [{ + "authority": "pool-0/set-0", + "canonical_path": "ilm/tier-delete-journal/legacy.json", + "etag": "etag-a", + "encoded_len": 6, + "content_sha256": content_sha256, + }] + }, + "topology_generation": hex_sha256(b"topology", ToOwned::to_owned), + "member_epochs_sha256": hex_sha256(b"members", ToOwned::to_owned), + })) + .unwrap(); + let payload = IlmRecoveryObservationReceipt { + schema: "rustfs-ilm-recovery-observation-receipt-v1".to_string(), + action: "export".to_string(), + actor_sha256: hex_sha256(b"actor-a", ToOwned::to_owned), + issued_at_unix_nanos: 1, + expires_at_unix_nanos: 1 + ILM_RECOVERY_OBSERVATION_RECEIPT_TTL.whole_nanoseconds() as i64, + nonce: Uuid::new_v4(), + observation, + }; + let credentials = Credentials { + access_key: "root".to_string(), + secret_key: "secret".to_string(), + ..Default::default() + }; + let token = encode_recovery_receipt(&payload, &credentials).unwrap(); + assert!(!token.contains("actor-a")); + assert!(!token.contains("ilm/tier-delete-journal")); + assert_eq!(decode_recovery_receipt(&token, &credentials).unwrap(), payload); + assert!( + validate_recovery_observation_receipt( + payload.clone(), + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ) + .is_ok() + ); + let assert_denied = |receipt: IlmRecoveryObservationReceipt, actor: &str, control: &str, now: i64| { + let err = validate_recovery_observation_receipt(receipt, actor, control, now) + .expect_err("invalid observation receipt must be denied"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + }; + assert_denied( + payload.clone(), + &hex_sha256(b"actor-b", ToOwned::to_owned), + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + assert_denied(payload.clone(), &payload.actor_sha256, &"cd".repeat(32), payload.issued_at_unix_nanos); + assert_denied( + payload.clone(), + &payload.actor_sha256, + &payload.observation.control_id, + payload.expires_at_unix_nanos, + ); + + let mut invalid = payload.clone(); + invalid.schema = "rustfs-ilm-recovery-observation-receipt-v2".to_string(); + assert_denied( + invalid, + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + let mut invalid = payload.clone(); + invalid.action = "abandon".to_string(); + assert_denied( + invalid, + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + let mut invalid = payload.clone(); + invalid.nonce = Uuid::nil(); + assert_denied( + invalid, + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + let mut invalid = payload.clone(); + invalid.issued_at_unix_nanos += 1; + invalid.expires_at_unix_nanos += 1; + assert_denied( + invalid, + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + let mut invalid = payload.clone(); + invalid.expires_at_unix_nanos += 1; + assert_denied( + invalid, + &payload.actor_sha256, + &payload.observation.control_id, + payload.issued_at_unix_nanos, + ); + + let mut tampered = token.into_bytes(); + let last = tampered.last_mut().unwrap(); + *last = if *last == b'a' { b'b' } else { b'a' }; + let err = decode_recovery_receipt(std::str::from_utf8(&tampered).unwrap(), &credentials) + .expect_err("tampered receipt must be denied"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn recovery_export_download_headers_prevent_caching_and_force_attachment() { + let export_id = "ab".repeat(32); + let headers = recovery_export_download_headers(&export_id, 123).unwrap(); + assert_eq!(headers.get(header::CONTENT_TYPE).unwrap(), "application/json"); + assert_eq!(headers.get(header::CACHE_CONTROL).unwrap(), "no-store"); + assert_eq!(headers.get(header::CONTENT_LENGTH).unwrap(), "123"); + assert_eq!( + headers.get(header::CONTENT_DISPOSITION).unwrap(), + &format!("attachment; filename=\"ilm-recovery-export-{export_id}.json\"") + ); + } + fn manual_transition_job_request(method: Method, path: &'static str) -> S3Request { S3Request { input: Body::empty(), diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 69c61768f..54d1715d6 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -507,6 +507,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ LIST_TIER, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/ilm/recovery/records/{control_id}", + SET_TIER, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/ilm/recovery/exports/{export_id}", + SET_TIER, + RouteRiskLevel::High, + ), admin(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER, RouteRiskLevel::High), admin( HttpMethod::Get, @@ -2173,6 +2185,8 @@ mod tests { fn route_policy_uses_tier_actions_for_transition_routes() { assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/recovery/records", LIST_TIER); assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/recovery/records/{control_id}", LIST_TIER); + assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/recovery/records/{control_id}", SET_TIER); + assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/recovery/exports/{export_id}", SET_TIER); assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER); assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SET_TIER); assert_action(HttpMethod::Delete, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SET_TIER); diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index eec928968..8da3b43e2 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -215,6 +215,16 @@ fn expected_admin_route_matrix() -> Vec { "/v3/ilm/recovery/records/{control_id}", "/v3/ilm/recovery/records/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ), + admin_route_sample( + Method::POST, + "/v3/ilm/recovery/records/{control_id}", + "/v3/ilm/recovery/records/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ), + admin_route_sample( + Method::GET, + "/v3/ilm/recovery/exports/{export_id}", + "/v3/ilm/recovery/exports/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + ), admin_route(Method::POST, "/v3/ilm/transition/run"), admin_route_sample( Method::GET, @@ -942,6 +952,16 @@ fn test_register_routes_cover_representative_admin_paths() { Method::GET, &admin_path("/v3/ilm/recovery/records/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), ); + assert_route( + &router, + Method::POST, + &admin_path("/v3/ilm/recovery/records/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + ); + assert_route( + &router, + Method::GET, + &admin_path("/v3/ilm/recovery/exports/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), + ); assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run")); assert_route( &router, diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 97e87c68c..c29159a47 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -233,7 +233,10 @@ pub(crate) mod lifecycle { super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunOptions; pub(crate) type ManualTransitionRunReport = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunReport; pub(crate) use super::ecstore_bucket::lifecycle::recovery_control::{ - IlmRecoveryClassification, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls, + IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls, + }; + pub(crate) use super::ecstore_bucket::lifecycle::recovery_export::{ + IlmRecoveryExportObservation, create_recovery_export, inspect_recovery_export_observation, load_recovery_export, }; pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{ TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator, diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 25b200fb2..dd44f134d 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -187,6 +187,30 @@ static NODE_CAPABILITY_SERVER_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4 // operations do not add another peer RPC. const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 4; +fn encode_heal_capability_response( + topology_member: &str, + remote_version_state_probe: bool, + recovery_export_probe: bool, +) -> Result, Status> { + if recovery_export_probe { + rustfs_protos::encode_remote_version_state_capability( + topology_member, + crate::storage::storage_api::ilm_recovery_export_local_process_epoch().as_bytes(), + ) + .map_err(|_| Status::internal("ILM recovery export capability length cannot be represented")) + } else if remote_version_state_probe { + rustfs_protos::encode_remote_version_state_capability(topology_member, NODE_CAPABILITY_SERVER_EPOCH.as_bytes()) + .map_err(|_| Status::internal("remote version state capability length cannot be represented")) + } else { + rustfs_protos::encode_cross_pool_fence_capability( + CROSS_POOL_FENCE_SUPPORTED_VERSION, + topology_member, + NODE_CAPABILITY_SERVER_EPOCH.as_bytes(), + ) + .map_err(|_| Status::internal("cross-pool fence capability length cannot be represented")) + } +} + fn admit_heal_control_replay( replay_cache: &mut HashMap>, request_id: &str, @@ -1003,7 +1027,8 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { } let remote_version_state_probe = rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command); let cross_pool_fence_probe = rustfs_protos::is_cross_pool_fence_capability_probe(&request.get_ref().command); - if remote_version_state_probe || cross_pool_fence_probe { + let recovery_export_probe = rustfs_protos::is_ilm_recovery_export_capability_probe(&request.get_ref().command); + if remote_version_state_probe || cross_pool_fence_probe || recovery_export_probe { let topology_member = self .endpoint_pools() .await @@ -1013,17 +1038,7 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { if topology_member.is_empty() { return Err(Status::failed_precondition("local topology member identity is unavailable")); } - let result = if remote_version_state_probe { - rustfs_protos::encode_remote_version_state_capability(&topology_member, NODE_CAPABILITY_SERVER_EPOCH.as_bytes()) - .map_err(|_| Status::internal("remote version state capability length cannot be represented"))? - } else { - rustfs_protos::encode_cross_pool_fence_capability( - CROSS_POOL_FENCE_SUPPORTED_VERSION, - &topology_member, - NODE_CAPABILITY_SERVER_EPOCH.as_bytes(), - ) - .map_err(|_| Status::internal("cross-pool fence capability length cannot be represented"))? - }; + let result = encode_heal_capability_response(&topology_member, remote_version_state_probe, recovery_export_probe)?; let canonical_response = rustfs_protos::canonical_heal_control_response_body( request.get_ref().version, &request.get_ref().topology_fingerprint, @@ -4040,6 +4055,19 @@ mod tests { .expect_err("proof from one challenge must not be reusable"); } + #[test] + fn ilm_recovery_export_probe_uses_the_shared_local_process_epoch() { + let result = super::encode_heal_capability_response("node-a:9000", false, true) + .expect("ILM recovery export capability should encode"); + let (member, epoch) = + rustfs_protos::decode_remote_version_state_capability(&result).expect("ILM recovery export capability should decode"); + assert_eq!(member, "node-a:9000"); + assert_eq!( + Uuid::from_slice(epoch).expect("capability epoch should be a UUID"), + crate::storage::storage_api::ilm_recovery_export_local_process_epoch(), + ); + } + #[tokio::test] async fn cross_pool_fence_probe_authenticates_supported_v4_state() { let _ = rustfs_credentials::set_global_rpc_secret("cross-pool-fence-node-service-test-secret".to_string()); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e35cf812d..886deb1cf 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -513,8 +513,8 @@ pub(crate) mod ecstore_notification { pub(crate) use rustfs_ecstore::api::notification::rotate_cross_pool_fence_fleet_proof_for_test; pub(crate) use rustfs_ecstore::api::notification::{ ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof, - cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys, - start_remote_version_state_fleet_probe, + cross_pool_fence_fleet_proof_matches, get_global_notification_sys, ilm_recovery_export_local_process_epoch, + new_global_notification_sys, start_remote_version_state_fleet_probe, }; } @@ -1182,6 +1182,10 @@ pub(crate) fn start_remote_version_state_fleet_probe(topology_fingerprint: Strin ecstore_notification::start_remote_version_state_fleet_probe(topology_fingerprint); } +pub(crate) fn ilm_recovery_export_local_process_epoch() -> uuid::Uuid { + ecstore_notification::ilm_recovery_export_local_process_epoch() +} + pub(crate) async fn read_config(api: Arc, file: &str) -> Result> { ecstore_config::com::read_config(api, file).await }