mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 12:09:12 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 950cce6f57 |
@@ -76,22 +76,6 @@ pub mod bucket {
|
||||
};
|
||||
}
|
||||
|
||||
pub mod recovery_disposition {
|
||||
pub use crate::bucket::lifecycle::recovery_disposition::{
|
||||
CreatedIlmRecoveryDisposition, IlmRecoveryDisposition, IlmRecoveryDispositionAction, IlmRecoveryDispositionError,
|
||||
IlmRecoveryDispositionIdentity, IlmRecoveryDispositionOwnerLease, IlmRecoveryDispositionReasonCode,
|
||||
IlmRecoveryDispositionState, ObservedIlmRecoveryDisposition, create_recovery_disposition_if_absent,
|
||||
load_recovery_disposition, recovery_disposition_id, save_recovery_disposition_if_current,
|
||||
};
|
||||
}
|
||||
|
||||
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,
|
||||
@@ -309,7 +293,7 @@ pub mod cache {
|
||||
pub mod capacity {
|
||||
pub use crate::core::pools::{
|
||||
DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
|
||||
path2_bucket_object, path2_bucket_object_with_base_path,
|
||||
is_pool_activation_fleet_proof_error, path2_bucket_object, path2_bucket_object_with_base_path,
|
||||
};
|
||||
pub use crate::store::utils::is_reserved_or_invalid_bucket;
|
||||
}
|
||||
@@ -462,12 +446,9 @@ 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, IlmRecoveryExportFleetProofToken,
|
||||
LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
|
||||
acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
|
||||
ClusterTierDailyStats, CrossPoolFenceFleetProofToken, LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr,
|
||||
NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_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,
|
||||
};
|
||||
|
||||
@@ -41,21 +41,6 @@ where
|
||||
com::read_config(api, file).await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
|
||||
where
|
||||
S: ObjectIO<
|
||||
Error = Error,
|
||||
RangeSpec = HTTPRangeSpec,
|
||||
HeaderMap = HeaderMap,
|
||||
ObjectOptions = ObjectOptions,
|
||||
ObjectInfo = ObjectInfo,
|
||||
GetObjectReader = GetObjectReader,
|
||||
PutObjectReader = PutObjReader,
|
||||
>,
|
||||
{
|
||||
com::read_config_limited_preserve_empty(api, file, max_bytes).await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_with_metadata<S>(api: Arc<S>, file: &str, opts: &ObjectOptions) -> Result<(Vec<u8>, ObjectInfo)>
|
||||
where
|
||||
S: ObjectIO<
|
||||
@@ -71,26 +56,6 @@ where
|
||||
com::read_config_with_metadata(api, file, opts).await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
|
||||
api: Arc<S>,
|
||||
file: &str,
|
||||
opts: &ObjectOptions,
|
||||
max_bytes: usize,
|
||||
) -> Result<(Vec<u8>, ObjectInfo)>
|
||||
where
|
||||
S: ObjectIO<
|
||||
Error = Error,
|
||||
RangeSpec = HTTPRangeSpec,
|
||||
HeaderMap = HeaderMap,
|
||||
ObjectOptions = ObjectOptions,
|
||||
ObjectInfo = ObjectInfo,
|
||||
GetObjectReader = GetObjectReader,
|
||||
PutObjectReader = PutObjReader,
|
||||
>,
|
||||
{
|
||||
com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
|
||||
}
|
||||
|
||||
pub(crate) async fn save_config<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()>
|
||||
where
|
||||
S: ObjectIO<
|
||||
|
||||
@@ -22,7 +22,7 @@ use super::{
|
||||
bucket_lifecycle_ops::{
|
||||
ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token,
|
||||
},
|
||||
manual_transition_job, recovery_control, recovery_disposition, recovery_export, tier_delete_journal, transition_transaction,
|
||||
manual_transition_job, recovery_control, tier_delete_journal, transition_transaction,
|
||||
};
|
||||
use crate::error::{Error, Result};
|
||||
use crate::services::tier::tier_probe_intent;
|
||||
@@ -42,8 +42,6 @@ pub(crate) enum DurableIlmRecordKind {
|
||||
ManualTransitionTask,
|
||||
ManualTransitionWorkerResult,
|
||||
RecoveryControl,
|
||||
RecoveryExport,
|
||||
RecoveryDisposition,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
@@ -114,20 +112,8 @@ 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 RECOVERY_DISPOSITION_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
|
||||
name: "recovery-disposition",
|
||||
prefix: recovery_disposition::ILM_RECOVERY_DISPOSITION_PREFIX,
|
||||
max_record_size: recovery_disposition::MAX_ILM_RECOVERY_DISPOSITION_SIZE,
|
||||
kind: DurableIlmRecordKind::RecoveryDisposition,
|
||||
};
|
||||
|
||||
pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 12] = [
|
||||
pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 10] = [
|
||||
TIER_DELETE_JOURNAL_NAMESPACE,
|
||||
TIER_DELETE_JOURNAL_V6_NAMESPACE,
|
||||
TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
|
||||
@@ -138,8 +124,6 @@ pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 12] = [
|
||||
MANUAL_TRANSITION_TASK_NAMESPACE,
|
||||
MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
|
||||
RECOVERY_CONTROL_NAMESPACE,
|
||||
RECOVERY_EXPORT_NAMESPACE,
|
||||
RECOVERY_DISPOSITION_NAMESPACE,
|
||||
];
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@@ -277,28 +261,6 @@ pub(crate) enum DurableIlmRecordCheckpoint {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
owner_fence_sha256: Option<String>,
|
||||
},
|
||||
RecoveryExport {
|
||||
content_sha256: String,
|
||||
source_generation_sha256: String,
|
||||
topology_generation: String,
|
||||
member_epochs_sha256: String,
|
||||
creator_sha256: String,
|
||||
retain_until_unix_nanos: i64,
|
||||
},
|
||||
RecoveryDisposition {
|
||||
content_sha256: String,
|
||||
identity_sha256: String,
|
||||
copy_manifest_sha256: String,
|
||||
copy_manifest_count: usize,
|
||||
created_at_unix_nanos: i64,
|
||||
revision: u64,
|
||||
state: recovery_disposition::IlmRecoveryDispositionState,
|
||||
owner_fence_sha256: Option<String>,
|
||||
owner_lease_acquired_at_unix_nanos: Option<i64>,
|
||||
owner_lease_expires_at_unix_nanos: Option<i64>,
|
||||
confirmed_absent_sha256: Vec<String>,
|
||||
retain_until_unix_nanos: i64,
|
||||
},
|
||||
}
|
||||
|
||||
impl DurableIlmRecordCheckpoint {
|
||||
@@ -313,9 +275,7 @@ impl DurableIlmRecordCheckpoint {
|
||||
| Self::ManualTransitionScope { content_sha256, .. }
|
||||
| Self::ManualTransitionTask { content_sha256 }
|
||||
| Self::ManualTransitionWorkerResult { content_sha256 }
|
||||
| Self::RecoveryControl { content_sha256, .. }
|
||||
| Self::RecoveryExport { content_sha256, .. }
|
||||
| Self::RecoveryDisposition { content_sha256, .. } => content_sha256,
|
||||
| Self::RecoveryControl { content_sha256, .. } => content_sha256,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -356,9 +316,6 @@ impl DurableIlmRecordCheckpoint {
|
||||
{
|
||||
return Err(Error::other("durable ILM tier delete journal checkpoint is invalid"));
|
||||
}
|
||||
if !recovery_disposition_checkpoint_is_valid(checkpoint) {
|
||||
return Err(Error::other("durable ILM recovery disposition checkpoint is invalid"));
|
||||
}
|
||||
}
|
||||
if self == next {
|
||||
if let Self::ManualTransitionJob {
|
||||
@@ -637,83 +594,6 @@ impl DurableIlmRecordCheckpoint {
|
||||
&& previous_attempts == next_attempts;
|
||||
adjacent && (claim || source_refresh || completion)
|
||||
}
|
||||
(
|
||||
Self::RecoveryDisposition {
|
||||
identity_sha256: previous_identity,
|
||||
copy_manifest_sha256: previous_manifest,
|
||||
copy_manifest_count: previous_manifest_count,
|
||||
created_at_unix_nanos: previous_created_at,
|
||||
revision: previous_revision,
|
||||
state: previous_state,
|
||||
owner_fence_sha256: previous_owner,
|
||||
owner_lease_acquired_at_unix_nanos: previous_owner_acquired,
|
||||
owner_lease_expires_at_unix_nanos: previous_owner_expires,
|
||||
confirmed_absent_sha256: previous_confirmed,
|
||||
retain_until_unix_nanos: previous_retain_until,
|
||||
..
|
||||
},
|
||||
Self::RecoveryDisposition {
|
||||
identity_sha256: next_identity,
|
||||
copy_manifest_sha256: next_manifest,
|
||||
copy_manifest_count: next_manifest_count,
|
||||
created_at_unix_nanos: next_created_at,
|
||||
revision: next_revision,
|
||||
state: next_state,
|
||||
owner_fence_sha256: next_owner,
|
||||
owner_lease_acquired_at_unix_nanos: next_owner_acquired,
|
||||
owner_lease_expires_at_unix_nanos: next_owner_expires,
|
||||
confirmed_absent_sha256: next_confirmed,
|
||||
retain_until_unix_nanos: next_retain_until,
|
||||
..
|
||||
},
|
||||
) => {
|
||||
use recovery_disposition::IlmRecoveryDispositionState::{Applying, Completed, Prepared};
|
||||
|
||||
let immutable_identity_matches = previous_identity == next_identity
|
||||
&& previous_manifest == next_manifest
|
||||
&& previous_manifest_count == next_manifest_count
|
||||
&& previous_created_at == next_created_at
|
||||
&& previous_retain_until == next_retain_until;
|
||||
let adjacent = previous_revision.checked_add(1) == Some(*next_revision);
|
||||
let progress_is_monotonic = sorted_sha256_set_is_subset(previous_confirmed, next_confirmed);
|
||||
let legal_edge = match (previous_state, next_state) {
|
||||
(Prepared, Prepared) => {
|
||||
let claim = previous_owner.is_none() && next_owner.is_some();
|
||||
let takeover = previous_owner.is_some()
|
||||
&& previous_owner != next_owner
|
||||
&& previous_owner_expires
|
||||
.zip(*next_owner_acquired)
|
||||
.is_some_and(|(expires, acquired)| acquired >= expires);
|
||||
previous_confirmed == next_confirmed && (claim || takeover)
|
||||
}
|
||||
(Prepared, Applying) => {
|
||||
previous_confirmed == next_confirmed
|
||||
&& previous_owner.is_some()
|
||||
&& previous_owner == next_owner
|
||||
&& previous_owner_acquired == next_owner_acquired
|
||||
&& previous_owner_expires == next_owner_expires
|
||||
}
|
||||
(Applying, Applying) => {
|
||||
let progress = previous_owner == next_owner
|
||||
&& previous_owner_acquired == next_owner_acquired
|
||||
&& previous_owner_expires == next_owner_expires
|
||||
&& previous_confirmed.len().checked_add(1) == Some(next_confirmed.len());
|
||||
let takeover = previous_owner.is_some()
|
||||
&& previous_owner != next_owner
|
||||
&& previous_confirmed == next_confirmed
|
||||
&& previous_owner_expires
|
||||
.zip(*next_owner_acquired)
|
||||
.is_some_and(|(expires, acquired)| acquired >= expires);
|
||||
progress || takeover
|
||||
}
|
||||
(Applying, Completed) => {
|
||||
previous_owner.is_some() && next_owner.is_none() && previous_confirmed == next_confirmed
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
|
||||
immutable_identity_matches && adjacent && progress_is_monotonic && legal_edge
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
|
||||
@@ -731,11 +611,6 @@ impl DurableIlmRecordCheckpoint {
|
||||
/// after the exact terminal ETag and terminal receipt were committed, to
|
||||
/// purge older object versions exposed by that deletion.
|
||||
pub(crate) fn is_predecessor_of_terminal(&self, terminal: &Self) -> bool {
|
||||
for checkpoint in [self, terminal] {
|
||||
if !recovery_disposition_checkpoint_is_valid(checkpoint) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
if let Self::TierProbeIntent { state, .. } = terminal
|
||||
&& !matches!(
|
||||
state,
|
||||
@@ -752,11 +627,6 @@ impl DurableIlmRecordCheckpoint {
|
||||
{
|
||||
return false;
|
||||
}
|
||||
if let Self::RecoveryDisposition { state, .. } = terminal
|
||||
&& state != &recovery_disposition::IlmRecoveryDispositionState::Completed
|
||||
{
|
||||
return false;
|
||||
}
|
||||
if self == terminal || self.validate_successor(terminal).is_ok() {
|
||||
return true;
|
||||
}
|
||||
@@ -882,121 +752,11 @@ impl DurableIlmRecordCheckpoint {
|
||||
&& terminal_revision > previous_revision
|
||||
&& terminal_attempts >= previous_attempts
|
||||
}
|
||||
(
|
||||
Self::RecoveryDisposition {
|
||||
identity_sha256: previous_identity,
|
||||
copy_manifest_sha256: previous_manifest,
|
||||
copy_manifest_count: previous_manifest_count,
|
||||
created_at_unix_nanos: previous_created_at,
|
||||
revision: previous_revision,
|
||||
state: previous_state,
|
||||
owner_fence_sha256: previous_owner,
|
||||
confirmed_absent_sha256: previous_confirmed,
|
||||
retain_until_unix_nanos: previous_retain_until,
|
||||
..
|
||||
},
|
||||
Self::RecoveryDisposition {
|
||||
identity_sha256: terminal_identity,
|
||||
copy_manifest_sha256: terminal_manifest,
|
||||
copy_manifest_count: terminal_manifest_count,
|
||||
created_at_unix_nanos: terminal_created_at,
|
||||
revision: terminal_revision,
|
||||
state: recovery_disposition::IlmRecoveryDispositionState::Completed,
|
||||
confirmed_absent_sha256: terminal_confirmed,
|
||||
retain_until_unix_nanos: terminal_retain_until,
|
||||
..
|
||||
},
|
||||
) => {
|
||||
matches!(
|
||||
previous_state,
|
||||
recovery_disposition::IlmRecoveryDispositionState::Prepared
|
||||
| recovery_disposition::IlmRecoveryDispositionState::Applying
|
||||
) && previous_identity == terminal_identity
|
||||
&& previous_manifest == terminal_manifest
|
||||
&& previous_manifest_count == terminal_manifest_count
|
||||
&& previous_created_at == terminal_created_at
|
||||
&& previous_retain_until == terminal_retain_until
|
||||
&& terminal_revision.checked_sub(*previous_revision).is_some_and(|distance| {
|
||||
let minimum_distance = match previous_state {
|
||||
recovery_disposition::IlmRecoveryDispositionState::Prepared if previous_owner.is_some() => 3,
|
||||
recovery_disposition::IlmRecoveryDispositionState::Prepared => 4,
|
||||
recovery_disposition::IlmRecoveryDispositionState::Applying
|
||||
if previous_confirmed.len() == *previous_manifest_count =>
|
||||
{
|
||||
1
|
||||
}
|
||||
recovery_disposition::IlmRecoveryDispositionState::Applying => 2,
|
||||
recovery_disposition::IlmRecoveryDispositionState::Completed => u64::MAX,
|
||||
};
|
||||
distance >= minimum_distance
|
||||
})
|
||||
&& sorted_sha256_set_is_subset(previous_confirmed, terminal_confirmed)
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn recovery_disposition_checkpoint_is_valid(checkpoint: &DurableIlmRecordCheckpoint) -> bool {
|
||||
use recovery_disposition::IlmRecoveryDispositionState::{Applying, Completed, Prepared};
|
||||
|
||||
let DurableIlmRecordCheckpoint::RecoveryDisposition {
|
||||
content_sha256,
|
||||
identity_sha256,
|
||||
copy_manifest_sha256,
|
||||
copy_manifest_count,
|
||||
created_at_unix_nanos,
|
||||
revision,
|
||||
state,
|
||||
owner_fence_sha256,
|
||||
owner_lease_acquired_at_unix_nanos,
|
||||
owner_lease_expires_at_unix_nanos,
|
||||
confirmed_absent_sha256,
|
||||
retain_until_unix_nanos,
|
||||
} = checkpoint
|
||||
else {
|
||||
return true;
|
||||
};
|
||||
let owner_fence_sha256 = owner_fence_sha256.as_deref();
|
||||
|
||||
is_canonical_sha256(content_sha256)
|
||||
&& is_canonical_sha256(identity_sha256)
|
||||
&& is_canonical_sha256(copy_manifest_sha256)
|
||||
&& *copy_manifest_count > 0
|
||||
&& *created_at_unix_nanos > 0
|
||||
&& *revision > 0
|
||||
&& *retain_until_unix_nanos > 0
|
||||
&& owner_fence_sha256.is_none_or(is_canonical_sha256)
|
||||
&& match (
|
||||
owner_fence_sha256,
|
||||
*owner_lease_acquired_at_unix_nanos,
|
||||
*owner_lease_expires_at_unix_nanos,
|
||||
) {
|
||||
(None, None, None) => true,
|
||||
(Some(_), Some(acquired), Some(expires)) => acquired > 0 && expires > acquired,
|
||||
_ => false,
|
||||
}
|
||||
&& confirmed_absent_sha256.len() <= *copy_manifest_count
|
||||
&& confirmed_absent_sha256.iter().all(|digest| is_canonical_sha256(digest))
|
||||
&& confirmed_absent_sha256.windows(2).all(|pair| pair[0] < pair[1])
|
||||
&& match *state {
|
||||
Prepared => confirmed_absent_sha256.is_empty(),
|
||||
Applying => owner_fence_sha256.is_some(),
|
||||
Completed => owner_fence_sha256.is_none() && confirmed_absent_sha256.len() == *copy_manifest_count,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_canonical_sha256(value: &str) -> bool {
|
||||
is_sha256_checksum(value)
|
||||
&& !value
|
||||
.bytes()
|
||||
.any(|byte| byte.is_ascii_hexdigit() && byte.is_ascii_uppercase())
|
||||
}
|
||||
|
||||
fn sorted_sha256_set_is_subset(subset: &[String], superset: &[String]) -> bool {
|
||||
subset.iter().all(|candidate| superset.binary_search(candidate).is_ok())
|
||||
}
|
||||
|
||||
fn tier_delete_dispatch_parent_progress_delta(
|
||||
previous_sequence: u64,
|
||||
previous_completed_journals: u64,
|
||||
@@ -1588,55 +1348,6 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
|
||||
},
|
||||
)
|
||||
}
|
||||
DurableIlmRecordKind::RecoveryExport => {
|
||||
let (protocol, export_id) = recovery_export::recovery_export_id_from_record_object_name(path)?;
|
||||
let export = recovery_export::IlmRecoveryExport::decode(&export_id, data)?;
|
||||
let canonical = recovery_export::recovery_export_record_object_name(protocol, &export_id)?;
|
||||
if canonical != path || export.protocol != protocol {
|
||||
return Err(Error::other("ILM recovery export path is not canonical"));
|
||||
}
|
||||
let source_generation_sha256 = checkpoint_hash(&export.source_generation)?;
|
||||
(
|
||||
"export_id",
|
||||
export_id,
|
||||
DurableIlmRecordCheckpoint::RecoveryExport {
|
||||
content_sha256,
|
||||
source_generation_sha256,
|
||||
topology_generation: export.topology_generation,
|
||||
member_epochs_sha256: export.member_epochs_sha256,
|
||||
creator_sha256: export.creator_sha256,
|
||||
retain_until_unix_nanos: export.retain_until_unix_nanos,
|
||||
},
|
||||
)
|
||||
}
|
||||
DurableIlmRecordKind::RecoveryDisposition => {
|
||||
// The disposition module owns strict schema, checksum, canonical
|
||||
// path, immutable-manifest, and state-specific validation. Keep
|
||||
// this boundary limited to decommission identity/checkpoint
|
||||
// projection so the two readers cannot accept different records.
|
||||
let disposition = recovery_disposition::decode_recovery_disposition_checkpoint(path, data)?;
|
||||
if disposition.content_sha256 != content_sha256 {
|
||||
return Err(Error::other("ILM recovery disposition checkpoint content digest is invalid"));
|
||||
}
|
||||
(
|
||||
"disposition_id",
|
||||
disposition.disposition_id,
|
||||
DurableIlmRecordCheckpoint::RecoveryDisposition {
|
||||
content_sha256: disposition.content_sha256,
|
||||
identity_sha256: disposition.identity_sha256,
|
||||
copy_manifest_sha256: disposition.copy_manifest_sha256,
|
||||
copy_manifest_count: disposition.copy_manifest_count,
|
||||
created_at_unix_nanos: disposition.created_at_unix_nanos,
|
||||
revision: disposition.revision,
|
||||
state: disposition.state,
|
||||
owner_fence_sha256: disposition.owner_fence_sha256,
|
||||
owner_lease_acquired_at_unix_nanos: disposition.owner_lease_acquired_at_unix_nanos,
|
||||
owner_lease_expires_at_unix_nanos: disposition.owner_lease_expires_at_unix_nanos,
|
||||
confirmed_absent_sha256: disposition.confirmed_absent_sha256,
|
||||
retain_until_unix_nanos: disposition.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()))?;
|
||||
@@ -1792,203 +1503,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn recovery_disposition_checkpoint(
|
||||
revision: u64,
|
||||
state: recovery_disposition::IlmRecoveryDispositionState,
|
||||
owner_fence: Option<&str>,
|
||||
confirmed_absent_sha256: Vec<String>,
|
||||
) -> DurableIlmRecordCheckpoint {
|
||||
let (owner_lease_acquired_at_unix_nanos, owner_lease_expires_at_unix_nanos) = match owner_fence {
|
||||
Some("f") => (Some(10), Some(20)),
|
||||
Some(_) => (Some(1), Some(10)),
|
||||
None => (None, None),
|
||||
};
|
||||
DurableIlmRecordCheckpoint::RecoveryDisposition {
|
||||
content_sha256: format!("{revision:064x}"),
|
||||
identity_sha256: "a".repeat(64),
|
||||
copy_manifest_sha256: "d".repeat(64),
|
||||
copy_manifest_count: 2,
|
||||
created_at_unix_nanos: 1_700_000_000_000_000_000,
|
||||
revision,
|
||||
state,
|
||||
owner_fence_sha256: owner_fence.map(|digest| digest.repeat(64)),
|
||||
owner_lease_acquired_at_unix_nanos,
|
||||
owner_lease_expires_at_unix_nanos,
|
||||
confirmed_absent_sha256,
|
||||
retain_until_unix_nanos: 1_820_000_000_000_000_000,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_disposition_namespace_is_registered_without_shadowing_its_root() {
|
||||
let disposition_id = "a".repeat(64);
|
||||
let path = format!(
|
||||
"{}/tier_delete_journal/{}/{}/{}.json",
|
||||
recovery_disposition::ILM_RECOVERY_DISPOSITION_PREFIX,
|
||||
&disposition_id[..2],
|
||||
&disposition_id[2..4],
|
||||
disposition_id
|
||||
);
|
||||
let namespace = classify_durable_ilm_record(&path)
|
||||
.expect("recovery disposition path should classify")
|
||||
.expect("recovery disposition should be durable");
|
||||
|
||||
assert_eq!(namespace, &RECOVERY_DISPOSITION_NAMESPACE);
|
||||
assert!(classify_durable_ilm_record(recovery_disposition::ILM_RECOVERY_DISPOSITION_PREFIX).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_disposition_checkpoint_accepts_only_monotonic_progress() {
|
||||
use recovery_disposition::IlmRecoveryDispositionState::{Applying, Completed, Prepared};
|
||||
|
||||
let first_copy = "b".repeat(64);
|
||||
let second_copy = "c".repeat(64);
|
||||
let prepared = recovery_disposition_checkpoint(1, Prepared, None, Vec::new());
|
||||
let claimed = recovery_disposition_checkpoint(2, Prepared, Some("e"), Vec::new());
|
||||
let applying = recovery_disposition_checkpoint(3, Applying, Some("e"), Vec::new());
|
||||
let first_absent = recovery_disposition_checkpoint(4, Applying, Some("e"), vec![first_copy.clone()]);
|
||||
let taken_over = recovery_disposition_checkpoint(5, Applying, Some("f"), vec![first_copy.clone()]);
|
||||
let all_absent = recovery_disposition_checkpoint(6, Applying, Some("f"), vec![first_copy.clone(), second_copy.clone()]);
|
||||
let completed = recovery_disposition_checkpoint(7, Completed, None, vec![first_copy.clone(), second_copy.clone()]);
|
||||
|
||||
prepared
|
||||
.validate_successor(&claimed)
|
||||
.expect("Prepared should record an owner claim without absence progress");
|
||||
claimed
|
||||
.validate_successor(&applying)
|
||||
.expect("Prepared should advance to Applying without folding in deletion progress");
|
||||
applying
|
||||
.validate_successor(&first_absent)
|
||||
.expect("Applying should append newly confirmed absent copies");
|
||||
first_absent
|
||||
.validate_successor(&taken_over)
|
||||
.expect("Applying should record a fenced owner takeover without losing progress");
|
||||
taken_over
|
||||
.validate_successor(&all_absent)
|
||||
.expect("Applying should preserve every earlier confirmation while making progress");
|
||||
all_absent
|
||||
.validate_successor(&completed)
|
||||
.expect("a fully confirmed manifest should advance to Completed");
|
||||
|
||||
assert!(
|
||||
prepared.validate_successor(&completed).is_err(),
|
||||
"adjacent receipt updates must not skip Applying"
|
||||
);
|
||||
assert!(
|
||||
first_absent
|
||||
.validate_successor(&recovery_disposition_checkpoint(5, Applying, Some("e"), Vec::new()))
|
||||
.is_err(),
|
||||
"confirmed-absent progress must not move backwards"
|
||||
);
|
||||
assert!(
|
||||
applying
|
||||
.validate_successor(&recovery_disposition_checkpoint(4, Completed, None, vec![first_copy.clone()]))
|
||||
.is_err(),
|
||||
"Completed must cover the complete immutable copy manifest"
|
||||
);
|
||||
assert!(
|
||||
completed
|
||||
.validate_successor(&recovery_disposition_checkpoint(7, Applying, Some("e"), vec![second_copy]))
|
||||
.is_err(),
|
||||
"Completed is terminal"
|
||||
);
|
||||
assert!(
|
||||
first_absent
|
||||
.validate_successor(&recovery_disposition_checkpoint(5, Applying, Some("e"), vec![first_copy.clone()]))
|
||||
.is_err(),
|
||||
"a same-state revision bump must change the owner fence or absence progress"
|
||||
);
|
||||
assert!(
|
||||
applying
|
||||
.validate_successor(&recovery_disposition_checkpoint(4, Applying, None, vec![first_copy.clone()]))
|
||||
.is_err(),
|
||||
"Applying must retain a fenced owner"
|
||||
);
|
||||
|
||||
let mut noncanonical_identity = claimed.clone();
|
||||
if let DurableIlmRecordCheckpoint::RecoveryDisposition { identity_sha256, .. } = &mut noncanonical_identity {
|
||||
*identity_sha256 = "A".repeat(64);
|
||||
}
|
||||
assert!(prepared.validate_successor(&noncanonical_identity).is_err());
|
||||
let mut changed_created_at = claimed.clone();
|
||||
if let DurableIlmRecordCheckpoint::RecoveryDisposition {
|
||||
created_at_unix_nanos, ..
|
||||
} = &mut changed_created_at
|
||||
{
|
||||
*created_at_unix_nanos += 1;
|
||||
}
|
||||
assert!(prepared.validate_successor(&changed_created_at).is_err());
|
||||
|
||||
let mut early_takeover = taken_over.clone();
|
||||
if let DurableIlmRecordCheckpoint::RecoveryDisposition {
|
||||
owner_lease_acquired_at_unix_nanos,
|
||||
..
|
||||
} = &mut early_takeover
|
||||
{
|
||||
*owner_lease_acquired_at_unix_nanos = Some(9);
|
||||
}
|
||||
assert!(first_absent.validate_successor(&early_takeover).is_err());
|
||||
assert!(
|
||||
applying
|
||||
.validate_successor(&recovery_disposition_checkpoint(
|
||||
4,
|
||||
Applying,
|
||||
Some("e"),
|
||||
vec!["c".repeat(64), "b".repeat(64)],
|
||||
))
|
||||
.is_err(),
|
||||
"confirmed-absent entries must be a canonical sorted set"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_disposition_terminal_predecessor_requires_exact_identity_and_full_manifest() {
|
||||
use recovery_disposition::IlmRecoveryDispositionState::{Applying, Completed, Prepared};
|
||||
|
||||
let first_copy = "b".repeat(64);
|
||||
let second_copy = "c".repeat(64);
|
||||
let prepared = recovery_disposition_checkpoint(1, Prepared, None, Vec::new());
|
||||
let applying = recovery_disposition_checkpoint(3, Applying, Some("e"), vec![first_copy.clone()]);
|
||||
let completed = recovery_disposition_checkpoint(5, Completed, None, vec![first_copy.clone(), second_copy]);
|
||||
|
||||
assert!(prepared.is_predecessor_of_terminal(&completed));
|
||||
assert!(applying.is_predecessor_of_terminal(&completed));
|
||||
assert!(
|
||||
!prepared.is_predecessor_of_terminal(&recovery_disposition_checkpoint(2, Applying, Some("e"), Vec::new())),
|
||||
"a nonterminal disposition must not authorize terminal cleanup"
|
||||
);
|
||||
assert!(
|
||||
!prepared.is_predecessor_of_terminal(&recovery_disposition_checkpoint(
|
||||
4,
|
||||
Completed,
|
||||
None,
|
||||
vec![first_copy.clone(), "c".repeat(64)],
|
||||
)),
|
||||
"terminal proof must leave enough revisions for claim, apply, progress, and completion"
|
||||
);
|
||||
assert!(
|
||||
!applying.is_predecessor_of_terminal(&recovery_disposition_checkpoint(
|
||||
4,
|
||||
Completed,
|
||||
None,
|
||||
vec![first_copy.clone(), "c".repeat(64)],
|
||||
)),
|
||||
"an incomplete Applying checkpoint cannot complete without a progress generation"
|
||||
);
|
||||
|
||||
let mut other_identity = completed.clone();
|
||||
if let DurableIlmRecordCheckpoint::RecoveryDisposition { identity_sha256, .. } = &mut other_identity {
|
||||
*identity_sha256 = "e".repeat(64);
|
||||
}
|
||||
assert!(!prepared.is_predecessor_of_terminal(&other_identity));
|
||||
|
||||
let incomplete_terminal = recovery_disposition_checkpoint(4, Completed, None, vec![first_copy]);
|
||||
assert!(
|
||||
!prepared.is_predecessor_of_terminal(&incomplete_terminal),
|
||||
"a partial confirmed-absent set must not become terminal proof"
|
||||
);
|
||||
}
|
||||
|
||||
fn tier_probe_intent_fixture() -> tier_probe_intent::TierProbeIntent {
|
||||
let probe_id = Uuid::parse_str("36e2220e-9ad2-495b-b3bc-c4d2caf70a31").expect("fixture uuid should parse");
|
||||
tier_probe_intent::TierProbeIntent {
|
||||
|
||||
@@ -25,8 +25,6 @@ mod object_handlers_common;
|
||||
mod object_lock_boundary;
|
||||
pub use self::core as lifecycle;
|
||||
pub mod recovery_control;
|
||||
pub mod recovery_disposition;
|
||||
pub mod recovery_export;
|
||||
mod replication_sink;
|
||||
pub mod rule;
|
||||
mod runtime_boundary;
|
||||
|
||||
@@ -168,7 +168,7 @@ impl IlmRecoverySourceGeneration {
|
||||
Ok(generation)
|
||||
}
|
||||
|
||||
pub(crate) fn validate(&self) -> Result<()> {
|
||||
fn validate(&self) -> Result<()> {
|
||||
if self.source_schema.trim().is_empty() {
|
||||
return Err(IlmRecoveryControlError::Corrupt("source schema is empty"));
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,840 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::{collections::HashSet, sync::Arc};
|
||||
|
||||
use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::config_boundary;
|
||||
use super::recovery_control::{
|
||||
IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryProtocol, IlmRecoverySourceCopy, IlmRecoverySourceGeneration,
|
||||
MAX_ILM_RECOVERY_CONTROL_SIZE, ObservedIlmRecoveryControl, ObservedIlmRecoverySource, recovery_control_record_object_name,
|
||||
};
|
||||
use super::tier_delete_journal::{
|
||||
TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA, validate_legacy_tier_delete_recovery_source,
|
||||
};
|
||||
use crate::disk::RUSTFS_META_BUCKET;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::object_api::{ObjectOptions, WriteCompletion};
|
||||
use crate::services::notification_sys::{
|
||||
acquire_ilm_recovery_export_fleet_proof, ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_member_epochs_sha256,
|
||||
ilm_recovery_export_topology_generation,
|
||||
};
|
||||
use crate::storage_api_contracts::{list::ListOperations as _, namespace::NamespaceLocking as _, object::HTTPPreconditions};
|
||||
use crate::store::ECStore;
|
||||
|
||||
pub const ILM_RECOVERY_EXPORT_SCHEMA: &str = "rustfs-ilm-recovery-export-v1";
|
||||
pub const ILM_RECOVERY_EXPORT_PREFIX: &str = "ilm/recovery-exports";
|
||||
pub const MAX_ILM_RECOVERY_EXPORT_SIZE: usize = 128 * 1024;
|
||||
const MAX_ILM_RECOVERY_EXPORTS: usize = 10_000;
|
||||
const MAX_ILM_RECOVERY_EXPORT_BYTES: u64 = 1024 * 1024 * 1024;
|
||||
const MAX_ACTOR_EXPORTS_PER_MINUTE: usize = 10;
|
||||
const MAX_CLUSTER_EXPORTS_PER_MINUTE: usize = 100;
|
||||
const EXPORT_RETENTION_NANOS: i64 = 90 * 24 * 60 * 60 * 1_000_000_000;
|
||||
const EXPORT_ADMISSION_LOCK: &str = "ilm/recovery-admission/export.lock";
|
||||
const MAX_LEGACY_TIER_DELETE_SOURCE_SIZE: usize = 64 * 1024;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct IlmRecoveryExportObservation {
|
||||
pub control_id: String,
|
||||
pub protocol: IlmRecoveryProtocol,
|
||||
pub control_etag: String,
|
||||
pub control_revision: u64,
|
||||
pub classification: IlmRecoveryClassification,
|
||||
pub canonical_source_path: String,
|
||||
pub source_generation: IlmRecoverySourceGeneration,
|
||||
pub topology_generation: String,
|
||||
pub member_epochs_sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct IlmRecoveryExport {
|
||||
pub export_id: String,
|
||||
pub control_id: String,
|
||||
pub protocol: IlmRecoveryProtocol,
|
||||
pub control_etag: String,
|
||||
pub control_revision: u64,
|
||||
pub classification: IlmRecoveryClassification,
|
||||
pub canonical_source_path: String,
|
||||
pub source_generation: IlmRecoverySourceGeneration,
|
||||
pub topology_generation: String,
|
||||
pub member_epochs_sha256: String,
|
||||
pub creator_sha256: String,
|
||||
pub created_at_unix_nanos: i64,
|
||||
pub retain_until_unix_nanos: i64,
|
||||
pub source_bytes_base64: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct PersistedIlmRecoveryExport {
|
||||
schema: String,
|
||||
content_sha256: String,
|
||||
export: IlmRecoveryExport,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct IlmRecoveryExportCreated {
|
||||
pub export_id: String,
|
||||
pub content_sha256: String,
|
||||
pub encoded: Vec<u8>,
|
||||
pub replayed: bool,
|
||||
}
|
||||
|
||||
impl IlmRecoveryExport {
|
||||
fn validate(&self) -> Result<()> {
|
||||
self.source_generation.validate().map_err(Error::other)?;
|
||||
validate_sha256(&self.export_id, "ILM recovery export ID is invalid")?;
|
||||
validate_sha256(&self.control_id, "ILM recovery export control ID is invalid")?;
|
||||
validate_sha256(&self.topology_generation, "ILM recovery export topology generation is invalid")?;
|
||||
validate_sha256(&self.member_epochs_sha256, "ILM recovery export member epoch digest is invalid")?;
|
||||
validate_sha256(&self.creator_sha256, "ILM recovery export creator digest is invalid")?;
|
||||
if self.protocol != IlmRecoveryProtocol::TierDeleteJournal
|
||||
|| self.classification != IlmRecoveryClassification::RetainedAmbiguous
|
||||
|| !is_legacy_export_schema(&self.source_generation.source_schema)
|
||||
{
|
||||
return Err(Error::other("ILM recovery export source is not an exportable legacy journal"));
|
||||
}
|
||||
if self.control_etag.trim().is_empty() || self.control_revision == 0 {
|
||||
return Err(Error::other("ILM recovery export control generation is invalid"));
|
||||
}
|
||||
if self.canonical_source_path.is_empty()
|
||||
|| self.canonical_source_path.starts_with('/')
|
||||
|| self.canonical_source_path.ends_with('/')
|
||||
|| self.canonical_source_path.split('/').any(str::is_empty)
|
||||
{
|
||||
return Err(Error::other("ILM recovery export source path is invalid"));
|
||||
}
|
||||
if self.created_at_unix_nanos <= 0
|
||||
|| self.retain_until_unix_nanos < self.created_at_unix_nanos.saturating_add(EXPORT_RETENTION_NANOS)
|
||||
{
|
||||
return Err(Error::other("ILM recovery export retention is invalid"));
|
||||
}
|
||||
let source = base64_simd::STANDARD
|
||||
.decode_to_vec(self.source_bytes_base64.as_bytes())
|
||||
.map_err(|_| Error::other("ILM recovery export source encoding is invalid"))?;
|
||||
validate_legacy_tier_delete_recovery_source(&self.canonical_source_path, &self.source_generation.source_schema, &source)?;
|
||||
let encoded_len = u64::try_from(source.len()).map_err(|_| Error::other("ILM recovery export source length overflow"))?;
|
||||
if source.is_empty()
|
||||
|| source.len() > MAX_LEGACY_TIER_DELETE_SOURCE_SIZE
|
||||
|| hex_sha256(&source, ToOwned::to_owned) != self.source_generation.content_sha256
|
||||
|| self.source_generation.copies.iter().any(|copy| {
|
||||
copy.canonical_path != self.canonical_source_path
|
||||
|| copy.etag != self.source_generation.source_etag
|
||||
|| copy.content_sha256 != self.source_generation.content_sha256
|
||||
|| copy.encoded_len != encoded_len
|
||||
})
|
||||
{
|
||||
return Err(Error::other("ILM recovery export source bytes do not match the observed generation"));
|
||||
}
|
||||
if recovery_export_id(&self.control_id, &self.source_generation)? != self.export_id {
|
||||
return Err(Error::other("ILM recovery export ID does not match its source generation"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn encode(&self) -> Result<Vec<u8>> {
|
||||
self.validate()?;
|
||||
let export_bytes = serde_json::to_vec(self).map_err(Error::other)?;
|
||||
let persisted = PersistedIlmRecoveryExport {
|
||||
schema: ILM_RECOVERY_EXPORT_SCHEMA.to_string(),
|
||||
content_sha256: hex_sha256(&export_bytes, ToOwned::to_owned),
|
||||
export: self.clone(),
|
||||
};
|
||||
let encoded = serde_json::to_vec(&persisted).map_err(Error::other)?;
|
||||
if encoded.len() > MAX_ILM_RECOVERY_EXPORT_SIZE {
|
||||
return Err(Error::other("encoded ILM recovery export exceeds maximum size"));
|
||||
}
|
||||
Ok(encoded)
|
||||
}
|
||||
|
||||
pub fn decode(expected_export_id: &str, data: &[u8]) -> Result<Self> {
|
||||
validate_sha256(expected_export_id, "ILM recovery export ID is invalid")?;
|
||||
if data.len() > MAX_ILM_RECOVERY_EXPORT_SIZE {
|
||||
return Err(Error::other("encoded ILM recovery export exceeds maximum size"));
|
||||
}
|
||||
let persisted: PersistedIlmRecoveryExport = serde_json::from_slice(data).map_err(Error::other)?;
|
||||
if persisted.schema != ILM_RECOVERY_EXPORT_SCHEMA {
|
||||
return Err(Error::other("ILM recovery export schema is unsupported"));
|
||||
}
|
||||
validate_sha256(&persisted.content_sha256, "ILM recovery export checksum is invalid")?;
|
||||
let export_bytes = serde_json::to_vec(&persisted.export).map_err(Error::other)?;
|
||||
if hex_sha256(&export_bytes, ToOwned::to_owned) != persisted.content_sha256 {
|
||||
return Err(Error::other("ILM recovery export checksum mismatch"));
|
||||
}
|
||||
persisted.export.validate()?;
|
||||
if persisted.export.export_id != expected_export_id {
|
||||
return Err(Error::other("ILM recovery export ID does not match record key"));
|
||||
}
|
||||
Ok(persisted.export)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn recovery_export_record_object_name(protocol: IlmRecoveryProtocol, export_id: &str) -> Result<String> {
|
||||
validate_sha256(export_id, "ILM recovery export ID is invalid")?;
|
||||
Ok(format!(
|
||||
"{}/{}/{}/{}/{}.json",
|
||||
ILM_RECOVERY_EXPORT_PREFIX,
|
||||
protocol.as_str(),
|
||||
&export_id[..2],
|
||||
&export_id[2..4],
|
||||
export_id
|
||||
))
|
||||
}
|
||||
|
||||
pub fn recovery_export_id_from_record_object_name(object: &str) -> Result<(IlmRecoveryProtocol, String)> {
|
||||
let suffix = object
|
||||
.strip_prefix(ILM_RECOVERY_EXPORT_PREFIX)
|
||||
.and_then(|suffix| suffix.strip_prefix('/'))
|
||||
.ok_or_else(|| Error::other("ILM recovery export path has wrong prefix"))?;
|
||||
let mut parts = suffix.split('/');
|
||||
let protocol = match parts.next() {
|
||||
Some("tier_delete_journal") => IlmRecoveryProtocol::TierDeleteJournal,
|
||||
_ => return Err(Error::other("ILM recovery export protocol is invalid")),
|
||||
};
|
||||
let shard_a = parts
|
||||
.next()
|
||||
.ok_or_else(|| Error::other("ILM recovery export path is incomplete"))?;
|
||||
let shard_b = parts
|
||||
.next()
|
||||
.ok_or_else(|| Error::other("ILM recovery export path is incomplete"))?;
|
||||
let export_id = parts
|
||||
.next()
|
||||
.and_then(|name| name.strip_suffix(".json"))
|
||||
.ok_or_else(|| Error::other("ILM recovery export suffix is invalid"))?;
|
||||
if parts.next().is_some() {
|
||||
return Err(Error::other("ILM recovery export path is not canonical"));
|
||||
}
|
||||
validate_sha256(export_id, "ILM recovery export ID is invalid")?;
|
||||
if shard_a != &export_id[..2] || shard_b != &export_id[2..4] {
|
||||
return Err(Error::other("ILM recovery export shard does not match export ID"));
|
||||
}
|
||||
Ok((protocol, export_id.to_string()))
|
||||
}
|
||||
|
||||
pub async fn inspect_recovery_export_observation(api: Arc<ECStore>, control_id: &str) -> Result<IlmRecoveryExportObservation> {
|
||||
let proof = acquire_ilm_recovery_export_fleet_proof()
|
||||
.await
|
||||
.ok_or_else(|| Error::other("ILM recovery export fleet proof is unavailable"))?;
|
||||
let observed_control = load_exportable_control(api.clone(), control_id).await?;
|
||||
let observed_source = observe_export_source(
|
||||
api,
|
||||
&observed_control.control.identity.canonical_source_path,
|
||||
&observed_control.control.observed_source_generation.source_schema,
|
||||
)
|
||||
.await?;
|
||||
if !observed_source.is_consistent()
|
||||
|| observed_source.generation != observed_control.control.observed_source_generation
|
||||
|| !ilm_recovery_export_fleet_proof_matches(&proof).await
|
||||
{
|
||||
return Err(Error::other("ILM recovery export observation changed or is incomplete"));
|
||||
}
|
||||
Ok(IlmRecoveryExportObservation {
|
||||
control_id: control_id.to_string(),
|
||||
protocol: observed_control.control.identity.protocol,
|
||||
control_etag: observed_control.etag,
|
||||
control_revision: observed_control.control.revision,
|
||||
classification: observed_control.control.classification,
|
||||
canonical_source_path: observed_control.control.identity.canonical_source_path,
|
||||
source_generation: observed_source.generation,
|
||||
topology_generation: ilm_recovery_export_topology_generation(&proof),
|
||||
member_epochs_sha256: ilm_recovery_export_member_epochs_sha256(&proof),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn create_recovery_export(
|
||||
api: Arc<ECStore>,
|
||||
observation: &IlmRecoveryExportObservation,
|
||||
creator_sha256: &str,
|
||||
) -> Result<IlmRecoveryExportCreated> {
|
||||
validate_sha256(creator_sha256, "ILM recovery export creator digest is invalid")?;
|
||||
let lock = api.new_ns_lock(RUSTFS_META_BUCKET, EXPORT_ADMISSION_LOCK).await?;
|
||||
let admission_guard = lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
|
||||
|
||||
let proof = acquire_ilm_recovery_export_fleet_proof()
|
||||
.await
|
||||
.ok_or_else(|| Error::other("ILM recovery export fleet proof is unavailable"))?;
|
||||
if ilm_recovery_export_topology_generation(&proof) != observation.topology_generation
|
||||
|| ilm_recovery_export_member_epochs_sha256(&proof) != observation.member_epochs_sha256
|
||||
{
|
||||
return Err(Error::PreconditionFailed);
|
||||
}
|
||||
let control_object = recovery_control_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, &observation.control_id)
|
||||
.map_err(Error::other)?;
|
||||
let control_lock = api.new_ns_lock(RUSTFS_META_BUCKET, &control_object).await?;
|
||||
let control_guard = control_lock
|
||||
.get_read_lock(crate::set_disk::get_lock_acquire_timeout())
|
||||
.await?;
|
||||
let source_lock = api
|
||||
.new_ns_lock(RUSTFS_META_BUCKET, &observation.canonical_source_path)
|
||||
.await?;
|
||||
let source_guard = source_lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
|
||||
let locks_current = || !admission_guard.is_lock_lost() && !control_guard.is_lock_lost() && !source_guard.is_lock_lost();
|
||||
let (current, current_source_bytes) = current_observation_under_proof_no_lock(api.clone(), observation, &proof).await?;
|
||||
if ¤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 = recovery_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<ECStore>, export_id: &str) -> Result<IlmRecoveryExportCreated> {
|
||||
let (created, _) = load_recovery_export_decoded(api, export_id).await?;
|
||||
Ok(created)
|
||||
}
|
||||
|
||||
async fn load_recovery_export_decoded(
|
||||
api: Arc<ECStore>,
|
||||
export_id: &str,
|
||||
) -> Result<(IlmRecoveryExportCreated, IlmRecoveryExport)> {
|
||||
let object = recovery_export_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, export_id)?;
|
||||
let encoded = config_boundary::read_config_limited_preserve_empty(api, &object, MAX_ILM_RECOVERY_EXPORT_SIZE).await?;
|
||||
let export = IlmRecoveryExport::decode(export_id, &encoded)?;
|
||||
let content_sha256 = hex_sha256(&encoded, ToOwned::to_owned);
|
||||
Ok((
|
||||
IlmRecoveryExportCreated {
|
||||
export_id: export.export_id.clone(),
|
||||
content_sha256,
|
||||
encoded,
|
||||
replayed: false,
|
||||
},
|
||||
export,
|
||||
))
|
||||
}
|
||||
|
||||
impl IlmRecoveryExportCreated {
|
||||
fn with_replayed(mut self) -> Self {
|
||||
self.replayed = true;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_exportable_control(api: Arc<ECStore>, control_id: &str) -> Result<ObservedIlmRecoveryControl> {
|
||||
load_exportable_control_with_options(api, control_id, &ObjectOptions::default()).await
|
||||
}
|
||||
|
||||
async fn load_exportable_control_no_lock(api: Arc<ECStore>, control_id: &str) -> Result<ObservedIlmRecoveryControl> {
|
||||
load_exportable_control_with_options(
|
||||
api,
|
||||
control_id,
|
||||
&ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn load_exportable_control_with_options(
|
||||
api: Arc<ECStore>,
|
||||
control_id: &str,
|
||||
options: &ObjectOptions,
|
||||
) -> Result<ObservedIlmRecoveryControl> {
|
||||
let object = recovery_control_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, control_id).map_err(Error::other)?;
|
||||
let (data, metadata) =
|
||||
config_boundary::read_config_limited_preserve_empty_with_metadata(api, &object, options, MAX_ILM_RECOVERY_CONTROL_SIZE)
|
||||
.await?;
|
||||
let etag = metadata
|
||||
.etag
|
||||
.filter(|etag| !etag.trim().is_empty())
|
||||
.ok_or_else(|| Error::other("ILM recovery control is missing an ETag"))?;
|
||||
let control = IlmRecoveryControl::decode(control_id, &data).map_err(Error::other)?;
|
||||
if control.identity.protocol != IlmRecoveryProtocol::TierDeleteJournal
|
||||
|| control.classification != IlmRecoveryClassification::RetainedAmbiguous
|
||||
|| !is_legacy_export_schema(&control.observed_source_generation.source_schema)
|
||||
{
|
||||
return Err(Error::other("ILM recovery control is not exportable"));
|
||||
}
|
||||
Ok(ObservedIlmRecoveryControl { control, etag })
|
||||
}
|
||||
|
||||
async fn current_observation_under_proof_no_lock(
|
||||
api: Arc<ECStore>,
|
||||
expected: &IlmRecoveryExportObservation,
|
||||
proof: &crate::services::notification_sys::IlmRecoveryExportFleetProofToken,
|
||||
) -> Result<(IlmRecoveryExportObservation, Vec<u8>)> {
|
||||
let observed_control = load_exportable_control_no_lock(api.clone(), &expected.control_id).await?;
|
||||
let observed_source = observe_export_source_no_lock(
|
||||
api,
|
||||
&observed_control.control.identity.canonical_source_path,
|
||||
&observed_control.control.observed_source_generation.source_schema,
|
||||
)
|
||||
.await?;
|
||||
let source_bytes = observed_source
|
||||
.canonical_data
|
||||
.clone()
|
||||
.ok_or_else(|| Error::other("ILM recovery export source copies diverge"))?;
|
||||
if !observed_source.is_consistent()
|
||||
|| observed_source.generation != observed_control.control.observed_source_generation
|
||||
|| !ilm_recovery_export_fleet_proof_matches(proof).await
|
||||
{
|
||||
return Err(Error::PreconditionFailed);
|
||||
}
|
||||
Ok((
|
||||
IlmRecoveryExportObservation {
|
||||
control_id: expected.control_id.clone(),
|
||||
protocol: observed_control.control.identity.protocol,
|
||||
control_etag: observed_control.etag,
|
||||
control_revision: observed_control.control.revision,
|
||||
classification: observed_control.control.classification,
|
||||
canonical_source_path: observed_control.control.identity.canonical_source_path,
|
||||
source_generation: observed_source.generation,
|
||||
topology_generation: ilm_recovery_export_topology_generation(proof),
|
||||
member_epochs_sha256: ilm_recovery_export_member_epochs_sha256(proof),
|
||||
},
|
||||
source_bytes,
|
||||
))
|
||||
}
|
||||
|
||||
async fn observe_export_source(
|
||||
api: Arc<ECStore>,
|
||||
canonical_path: &str,
|
||||
source_schema: &str,
|
||||
) -> Result<ObservedIlmRecoverySource> {
|
||||
if canonical_path.is_empty()
|
||||
|| canonical_path.starts_with('/')
|
||||
|| canonical_path.ends_with('/')
|
||||
|| canonical_path.split('/').any(str::is_empty)
|
||||
|| !is_legacy_export_schema(source_schema)
|
||||
{
|
||||
return Err(Error::other("ILM recovery export source identity is invalid"));
|
||||
}
|
||||
let lock = api.new_ns_lock(RUSTFS_META_BUCKET, canonical_path).await?;
|
||||
let _guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
|
||||
observe_export_source_no_lock(api, canonical_path, source_schema).await
|
||||
}
|
||||
|
||||
async fn observe_export_source_no_lock(
|
||||
api: Arc<ECStore>,
|
||||
canonical_path: &str,
|
||||
source_schema: &str,
|
||||
) -> Result<ObservedIlmRecoverySource> {
|
||||
let mut copies = Vec::new();
|
||||
let mut canonical: Option<(String, String, Vec<u8>)> = None;
|
||||
let mut consistent = true;
|
||||
for set in api.all_set_disks() {
|
||||
let authority = format!("pool-{}/set-{}", set.pool_index, set.set_index);
|
||||
let result = config_boundary::read_config_limited_preserve_empty_with_metadata(
|
||||
set,
|
||||
canonical_path,
|
||||
&ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
},
|
||||
MAX_LEGACY_TIER_DELETE_SOURCE_SIZE,
|
||||
)
|
||||
.await;
|
||||
match result {
|
||||
Ok((data, metadata)) => {
|
||||
if data.is_empty() || data.len() > MAX_LEGACY_TIER_DELETE_SOURCE_SIZE {
|
||||
return Err(Error::other("ILM recovery export source exceeds its protocol size limit"));
|
||||
}
|
||||
validate_legacy_tier_delete_recovery_source(canonical_path, source_schema, &data)?;
|
||||
let etag = metadata
|
||||
.etag
|
||||
.filter(|etag| !etag.trim().is_empty())
|
||||
.ok_or_else(|| Error::other("ILM recovery export source copy is missing an ETag"))?;
|
||||
let content_sha256 = hex_sha256(&data, ToOwned::to_owned);
|
||||
let encoded_len =
|
||||
u64::try_from(data.len()).map_err(|_| Error::other("ILM recovery export source length does not fit u64"))?;
|
||||
copies.push(IlmRecoverySourceCopy {
|
||||
authority,
|
||||
canonical_path: canonical_path.to_string(),
|
||||
etag: etag.clone(),
|
||||
encoded_len,
|
||||
content_sha256: content_sha256.clone(),
|
||||
});
|
||||
match canonical.as_ref() {
|
||||
Some((first_etag, first_digest, first_data)) => {
|
||||
consistent &= first_etag == &etag && first_digest == &content_sha256 && first_data == &data;
|
||||
}
|
||||
None => canonical = Some((etag, content_sha256, data)),
|
||||
}
|
||||
}
|
||||
Err(err) if export_source_is_missing(&err) => {}
|
||||
Err(err) => return Err(err),
|
||||
}
|
||||
}
|
||||
let Some((source_etag, content_sha256, source_bytes)) = canonical else {
|
||||
return Err(Error::ConfigNotFound);
|
||||
};
|
||||
let generation =
|
||||
IlmRecoverySourceGeneration::new(source_schema, source_etag, content_sha256, copies).map_err(Error::other)?;
|
||||
Ok(ObservedIlmRecoverySource {
|
||||
generation,
|
||||
canonical_data: consistent.then_some(source_bytes),
|
||||
})
|
||||
}
|
||||
|
||||
fn export_source_is_missing(err: &Error) -> bool {
|
||||
matches!(
|
||||
err,
|
||||
Error::ConfigNotFound | Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::VersionNotFound(_, _, _)
|
||||
)
|
||||
}
|
||||
|
||||
fn build_export_from_source(
|
||||
observation: &IlmRecoveryExportObservation,
|
||||
creator_sha256: &str,
|
||||
created_at_unix_nanos: i64,
|
||||
source_bytes_base64: &str,
|
||||
) -> Result<IlmRecoveryExport> {
|
||||
let retain_until_unix_nanos = created_at_unix_nanos
|
||||
.checked_add(EXPORT_RETENTION_NANOS)
|
||||
.ok_or_else(|| Error::other("ILM recovery export retention timestamp overflow"))?;
|
||||
let export = IlmRecoveryExport {
|
||||
export_id: recovery_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)
|
||||
}
|
||||
|
||||
pub(crate) fn recovery_export_id(control_id: &str, generation: &IlmRecoverySourceGeneration) -> Result<String> {
|
||||
validate_sha256(control_id, "ILM recovery export control ID is invalid")?;
|
||||
validate_sha256(&generation.content_sha256, "ILM recovery export source checksum is invalid")?;
|
||||
validate_sha256(&generation.copy_set_sha256, "ILM recovery export copy-set checksum is invalid")?;
|
||||
let mut data = Vec::new();
|
||||
for part in [control_id, &generation.content_sha256, &generation.copy_set_sha256] {
|
||||
data.extend_from_slice(&(part.len() as u64).to_be_bytes());
|
||||
data.extend_from_slice(part.as_bytes());
|
||||
}
|
||||
Ok(hex_sha256(&data, ToOwned::to_owned))
|
||||
}
|
||||
|
||||
fn export_matches_observation(export: &IlmRecoveryExport, observation: &IlmRecoveryExportObservation) -> bool {
|
||||
export.control_id == observation.control_id
|
||||
&& export.protocol == observation.protocol
|
||||
&& export.classification == observation.classification
|
||||
&& export.canonical_source_path == observation.canonical_source_path
|
||||
&& export.source_generation == observation.source_generation
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
struct IlmRecoveryExportInventory {
|
||||
count: usize,
|
||||
bytes: u64,
|
||||
creations: Vec<(i64, String)>,
|
||||
}
|
||||
|
||||
impl IlmRecoveryExportInventory {
|
||||
fn check(&self, creator_sha256: &str, candidate_len: usize, now: i64) -> Result<()> {
|
||||
let recent_after = now.saturating_sub(60 * 1_000_000_000);
|
||||
let cluster_recent = self
|
||||
.creations
|
||||
.iter()
|
||||
.filter(|(created_at, _)| *created_at > recent_after)
|
||||
.count();
|
||||
let actor_recent = self
|
||||
.creations
|
||||
.iter()
|
||||
.filter(|(created_at, creator)| *created_at > recent_after && creator == creator_sha256)
|
||||
.count();
|
||||
check_export_admission(self.count, self.bytes, actor_recent, cluster_recent, candidate_len)
|
||||
}
|
||||
}
|
||||
|
||||
async fn collect_export_inventory(api: Arc<ECStore>) -> Result<IlmRecoveryExportInventory> {
|
||||
let mut marker = None;
|
||||
let mut seen_markers = HashSet::new();
|
||||
let mut inventory = IlmRecoveryExportInventory::default();
|
||||
loop {
|
||||
let page = api
|
||||
.clone()
|
||||
.list_objects_v2(
|
||||
RUSTFS_META_BUCKET,
|
||||
&format!("{ILM_RECOVERY_EXPORT_PREFIX}/"),
|
||||
marker.clone(),
|
||||
None,
|
||||
1_000,
|
||||
false,
|
||||
None,
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
for object in page.objects {
|
||||
let (_, export_id) = recovery_export_id_from_record_object_name(&object.name)?;
|
||||
let (stored, export) = load_recovery_export_decoded(api.clone(), &export_id).await?;
|
||||
inventory.count = inventory
|
||||
.count
|
||||
.checked_add(1)
|
||||
.ok_or_else(|| Error::other("ILM recovery export count overflow"))?;
|
||||
inventory.bytes = inventory
|
||||
.bytes
|
||||
.checked_add(u64::try_from(stored.encoded.len()).map_err(|_| Error::other("ILM recovery export size overflow"))?)
|
||||
.ok_or_else(|| Error::other("ILM recovery export byte total overflow"))?;
|
||||
inventory
|
||||
.creations
|
||||
.push((export.created_at_unix_nanos, export.creator_sha256));
|
||||
}
|
||||
if !page.is_truncated {
|
||||
break;
|
||||
}
|
||||
let next = page
|
||||
.next_continuation_token
|
||||
.ok_or_else(|| Error::other("ILM recovery export inventory omitted its continuation marker"))?;
|
||||
marker = Some(record_export_inventory_marker(&mut seen_markers, next)?);
|
||||
}
|
||||
Ok(inventory)
|
||||
}
|
||||
|
||||
fn record_export_inventory_marker(seen_markers: &mut HashSet<String>, next: String) -> Result<String> {
|
||||
if !seen_markers.insert(next.clone()) {
|
||||
return Err(Error::other("ILM recovery export inventory repeated its continuation marker"));
|
||||
}
|
||||
Ok(next)
|
||||
}
|
||||
|
||||
fn check_export_admission(
|
||||
count: usize,
|
||||
bytes: u64,
|
||||
actor_recent: usize,
|
||||
cluster_recent: usize,
|
||||
candidate_len: usize,
|
||||
) -> Result<()> {
|
||||
let candidate_len = u64::try_from(candidate_len).map_err(|_| Error::other("ILM recovery export size does not fit u64"))?;
|
||||
if count >= MAX_ILM_RECOVERY_EXPORTS
|
||||
|| bytes
|
||||
.checked_add(candidate_len)
|
||||
.is_none_or(|total| total > MAX_ILM_RECOVERY_EXPORT_BYTES)
|
||||
|| actor_recent >= MAX_ACTOR_EXPORTS_PER_MINUTE
|
||||
|| cluster_recent >= MAX_CLUSTER_EXPORTS_PER_MINUTE
|
||||
{
|
||||
return Err(Error::SlowDown);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn is_legacy_export_schema(schema: &str) -> bool {
|
||||
matches!(schema, TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA | TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA)
|
||||
}
|
||||
|
||||
fn validate_sha256(value: &str, message: &'static str) -> Result<()> {
|
||||
if !is_sha256_checksum(value)
|
||||
|| value
|
||||
.bytes()
|
||||
.any(|byte| byte.is_ascii_hexdigit() && byte.is_ascii_uppercase())
|
||||
{
|
||||
return Err(Error::other(message));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn now_unix_nanos() -> Result<i64> {
|
||||
i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos())
|
||||
.map_err(|_| Error::other("ILM recovery export timestamp does not fit i64"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::lifecycle::recovery_control::IlmRecoverySourceCopy;
|
||||
|
||||
const PINNED_V1_EXPORT: &[u8] = br#"{"schema":"rustfs-ilm-recovery-export-v1","content_sha256":"3dfb3ec3892256e909de1211c1a963ca7008963ff32b3a869f7161a7b9b44028","export":{"export_id":"2b78e7a825bfc2edbf7f773d0b6ed3bf93e360ff1702d73a449109c11bfaa105","control_id":"0fcd568a5cb9bdb4677b69354b11ee415af8f784519cff3da49a26f84eaee7f2","protocol":"tier_delete_journal","control_etag":"control-etag","control_revision":1,"classification":"retained_ambiguous","canonical_source_path":"ilm/tier-delete-journal/872072554f66ab326f10ce7adbae11422b7a4b0663aa7112d6061a8f6ed41b94.json","source_generation":{"source_schema":"rustfs-tier-delete-journal-v1","source_etag":"etag-a","content_sha256":"0e0b010ebdeeb7b41473fe8575e989d6bb1303c0ca551dd984e9400f0ae306bd","copy_set_sha256":"5a7406115b6c3923ffe79dcd1f43ccae7beed786e557163f019dd10ec409a653","copies":[{"authority":"pool-0/set-0","canonical_path":"ilm/tier-delete-journal/872072554f66ab326f10ce7adbae11422b7a4b0663aa7112d6061a8f6ed41b94.json","etag":"etag-a","encoded_len":81,"content_sha256":"0e0b010ebdeeb7b41473fe8575e989d6bb1303c0ca551dd984e9400f0ae306bd"}]},"topology_generation":"e6e2b826e31fca5c36125c48f130dcb6f961e698ff8a8776a1f290cf0892e8e6","member_epochs_sha256":"612dd8a861161819a4ad8f6f3e2a0567602877c043a2353ca933a13c78dc0ed4","creator_sha256":"50c9c4aeb40b5b206b6d98f516f8b8c0efd29ce2e56a76b345fb9240c225a1b7","created_at_unix_nanos":1000000000,"retain_until_unix_nanos":7776001000000000,"source_bytes_base64":"eyJ2ZXJzaW9uIjoxLCJvYmpfbmFtZSI6ImxlZ2FjeS9yZW1vdGUiLCJ2ZXJzaW9uX2lkIjoib3BhcXVlIiwidGllcl9uYW1lIjoiV0FSTSJ9"}}"#;
|
||||
|
||||
fn legacy_source() -> Vec<u8> {
|
||||
br#"{"version":1,"obj_name":"legacy/remote","version_id":"opaque","tier_name":"WARM"}"#.to_vec()
|
||||
}
|
||||
|
||||
fn observation() -> IlmRecoveryExportObservation {
|
||||
let source = legacy_source();
|
||||
let source_path = super::super::tier_delete_journal::tier_delete_journal_object_name(
|
||||
&super::super::tier_delete_journal::decode_tier_delete_journal_entry(&source).expect("legacy fixture should decode"),
|
||||
);
|
||||
let source_sha256 = hex_sha256(&source, ToOwned::to_owned);
|
||||
let generation = IlmRecoverySourceGeneration::new(
|
||||
TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA,
|
||||
"etag-a",
|
||||
source_sha256.clone(),
|
||||
vec![IlmRecoverySourceCopy {
|
||||
authority: "pool-0/set-0".to_string(),
|
||||
canonical_path: source_path.clone(),
|
||||
etag: "etag-a".to_string(),
|
||||
encoded_len: source.len() as u64,
|
||||
content_sha256: source_sha256,
|
||||
}],
|
||||
)
|
||||
.expect("generation should be valid");
|
||||
IlmRecoveryExportObservation {
|
||||
control_id: hex_sha256(b"control", ToOwned::to_owned),
|
||||
protocol: IlmRecoveryProtocol::TierDeleteJournal,
|
||||
control_etag: "control-etag".to_string(),
|
||||
control_revision: 1,
|
||||
classification: IlmRecoveryClassification::RetainedAmbiguous,
|
||||
canonical_source_path: source_path,
|
||||
source_generation: generation,
|
||||
topology_generation: hex_sha256(b"topology", ToOwned::to_owned),
|
||||
member_epochs_sha256: hex_sha256(b"epochs", ToOwned::to_owned),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_export_round_trip_is_strict_and_deterministic() {
|
||||
let observed = observation();
|
||||
let creator = hex_sha256(b"actor", ToOwned::to_owned);
|
||||
let export = build_export_from_source(
|
||||
&observed,
|
||||
&creator,
|
||||
1_000_000_000,
|
||||
&base64_simd::STANDARD.encode_to_string(legacy_source()),
|
||||
)
|
||||
.expect("export should be valid");
|
||||
assert_eq!(
|
||||
export.export_id,
|
||||
recovery_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());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
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_V1_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v1";
|
||||
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,23 +884,6 @@ struct PersistedTierDeleteJournalEntry {
|
||||
}
|
||||
|
||||
impl PersistedTierDeleteJournalEntry {
|
||||
fn validate_legacy_recovery_shape(&self) -> Result<()> {
|
||||
let has_later_version_fields = self.version_id_exact.is_some()
|
||||
|| self.version_state.is_some()
|
||||
|| self.state.is_some()
|
||||
|| self.source.is_some()
|
||||
|| self.dispatch.is_some();
|
||||
match self.version {
|
||||
1 if self.backend_identity.is_none() && !has_later_version_fields => Ok(()),
|
||||
TIER_DELETE_JOURNAL_VERSION if self.backend_identity.is_some() && !has_later_version_fields => Ok(()),
|
||||
1 => Err(Error::other("tier delete journal v1 entry contains fields from a later version")),
|
||||
TIER_DELETE_JOURNAL_VERSION => Err(Error::other(
|
||||
"tier delete journal v2 entry is missing its identity or contains fields from a later version",
|
||||
)),
|
||||
_ => Err(Error::other("tier delete journal is not an exportable legacy version")),
|
||||
}
|
||||
}
|
||||
|
||||
fn from_jentry(je: &Jentry) -> Result<Self> {
|
||||
validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?;
|
||||
let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown;
|
||||
@@ -5548,12 +5531,6 @@ fn canonical_legacy_tier_delete_journal_identity(object_name: &str) -> Option<&s
|
||||
.then_some(identity)
|
||||
}
|
||||
|
||||
pub(crate) fn validate_legacy_tier_delete_recovery_path(object_name: &str) -> Result<()> {
|
||||
canonical_legacy_tier_delete_journal_identity(object_name)
|
||||
.map(|_| ())
|
||||
.ok_or_else(|| Error::other("legacy tier delete journal path is not canonical"))
|
||||
}
|
||||
|
||||
fn legacy_tier_delete_recovery_descriptor(entry: &Jentry) -> Option<(&'static str, &'static str)> {
|
||||
match entry.persisted_version {
|
||||
1 => Some((TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V1_RECOVERY_CLASS)),
|
||||
@@ -5562,21 +5539,6 @@ 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<()> {
|
||||
validate_legacy_tier_delete_recovery_path(object_name)?;
|
||||
let persisted: PersistedTierDeleteJournalEntry =
|
||||
serde_json::from_slice(data).map_err(|err| Error::other_with_context("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,
|
||||
@@ -6178,18 +6140,17 @@ where
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
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,
|
||||
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,
|
||||
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, validate_legacy_tier_delete_recovery_source,
|
||||
tier_delete_source_matches_dispatch_scope,
|
||||
};
|
||||
use crate::bucket::lifecycle::tier_sweeper::{
|
||||
Jentry, TierDeleteDispatchBinding, TierDeleteJournalState, TierDeleteSourceIdentity,
|
||||
@@ -6648,72 +6609,6 @@ 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();
|
||||
|
||||
@@ -1688,15 +1688,6 @@ 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;
|
||||
|
||||
@@ -449,14 +449,6 @@ where
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
{
|
||||
let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
@@ -465,6 +457,14 @@ where
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
{
|
||||
let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
|
||||
api: Arc<S>,
|
||||
file: &str,
|
||||
@@ -476,18 +476,6 @@ where
|
||||
read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config_limited_preserve_empty_with_metadata_opts<S>(
|
||||
api: Arc<S>,
|
||||
file: &str,
|
||||
opts: &ObjectOptions,
|
||||
max_bytes: usize,
|
||||
) -> Result<(Vec<u8>, ObjectInfo)>
|
||||
where
|
||||
S: EcstoreObjectIO,
|
||||
{
|
||||
read_config_with_metadata_inner(api, file, opts, true, Some(max_bytes)).await
|
||||
}
|
||||
|
||||
/// Read an existing config object without treating an empty payload as absent.
|
||||
/// Callers that validate their own payload format need to distinguish corruption
|
||||
/// from `ConfigNotFound`.
|
||||
|
||||
@@ -3385,7 +3385,7 @@ pub(crate) async fn acquire_pool_activation_fleet_proof(
|
||||
.ok_or_else(|| Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED))
|
||||
}
|
||||
|
||||
pub(crate) fn is_pool_activation_fleet_proof_error(err: &Error) -> bool {
|
||||
pub fn is_pool_activation_fleet_proof_error(err: &Error) -> bool {
|
||||
// Save-stage helpers add context by formatting the original error, so the
|
||||
// marker may be nested in the display string. Restrict matching to the
|
||||
// `Error::other` I/O shape used by this activation path.
|
||||
|
||||
@@ -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, BTreeSet, HashMap, hash_map::DefaultHasher};
|
||||
use std::collections::{BTreeMap, HashMap, hash_map::DefaultHasher};
|
||||
use std::future::Future;
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::sync::{
|
||||
Arc, LazyLock, Mutex, OnceLock,
|
||||
Arc, Mutex, OnceLock,
|
||||
atomic::{AtomicBool, AtomicUsize, Ordering},
|
||||
};
|
||||
use std::time::{Duration, Instant, SystemTime};
|
||||
@@ -311,20 +311,12 @@ pub struct LegacyTransitionStateReconcileFleetProofToken {
|
||||
_permit: FleetCapabilityProofPermit,
|
||||
}
|
||||
|
||||
/// Effect-window authority for one immutable ILM recovery export.
|
||||
pub struct IlmRecoveryExportFleetProofToken {
|
||||
token: FleetCapabilityProofToken,
|
||||
_permit: FleetCapabilityProofPermit,
|
||||
}
|
||||
|
||||
static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static ILM_RECOVERY_EXPORT_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
|
||||
static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
|
||||
static ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
|
||||
|
||||
fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
|
||||
CROSS_POOL_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
|
||||
@@ -346,10 +338,6 @@ fn legacy_transition_state_reconcile_fleet_proof_slot() -> &'static std::sync::R
|
||||
LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
|
||||
}
|
||||
|
||||
fn ilm_recovery_export_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
|
||||
ILM_RECOVERY_EXPORT_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
|
||||
}
|
||||
|
||||
fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
|
||||
if let Some(proof) = state.proof.take() {
|
||||
proof.generation.revoke();
|
||||
@@ -585,117 +573,6 @@ pub async fn legacy_transition_state_reconcile_fleet_proof_matches(
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn acquire_ilm_recovery_export_fleet_proof() -> Option<IlmRecoveryExportFleetProofToken> {
|
||||
let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
|
||||
let proof = {
|
||||
let state = ilm_recovery_export_fleet_proof_slot()
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
acquire_ilm_recovery_export_fleet_proof_from(&state, expected_topology, Instant::now())?
|
||||
};
|
||||
let observed = observe_ilm_recovery_export_fleet(expected_topology).await?;
|
||||
let state = ilm_recovery_export_fleet_proof_slot()
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
ilm_recovery_export_fleet_proof_matches_observation_at(&state, &proof, expected_topology, &observed, Instant::now())
|
||||
.then_some(proof)
|
||||
}
|
||||
|
||||
fn acquire_ilm_recovery_export_fleet_proof_from(
|
||||
state: &FleetCapabilityProofState,
|
||||
expected_topology: &str,
|
||||
now: Instant,
|
||||
) -> Option<IlmRecoveryExportFleetProofToken> {
|
||||
let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
|
||||
let permit = state.proof.as_ref()?.generation.try_acquire()?;
|
||||
Some(IlmRecoveryExportFleetProofToken { token, _permit: permit })
|
||||
}
|
||||
|
||||
pub async fn ilm_recovery_export_fleet_proof_matches(proof: &IlmRecoveryExportFleetProofToken) -> bool {
|
||||
let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
|
||||
return false;
|
||||
};
|
||||
{
|
||||
let state = ilm_recovery_export_fleet_proof_slot()
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if !ilm_recovery_export_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
let Some(observed) = observe_ilm_recovery_export_fleet(expected_topology).await else {
|
||||
return false;
|
||||
};
|
||||
let state = ilm_recovery_export_fleet_proof_slot()
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
ilm_recovery_export_fleet_proof_matches_observation_at(&state, proof, expected_topology, &observed, Instant::now())
|
||||
}
|
||||
|
||||
pub fn ilm_recovery_export_topology_generation(proof: &IlmRecoveryExportFleetProofToken) -> String {
|
||||
let mut hasher = Sha256::new();
|
||||
hasher.update(b"rustfs-ilm-recovery-export-topology-v1\0");
|
||||
hasher.update(proof.token.topology_fingerprint.as_bytes());
|
||||
rustfs_utils::crypto::hex(hasher.finalize().as_slice())
|
||||
}
|
||||
|
||||
pub fn ilm_recovery_export_member_epochs_sha256(proof: &IlmRecoveryExportFleetProofToken) -> String {
|
||||
let encoded = serde_json::to_vec(proof.token.peer_epochs.as_ref()).expect("member epoch map is JSON encodable");
|
||||
let mut hasher = Sha256::new();
|
||||
hasher.update(b"rustfs-ilm-recovery-export-members-v1\0");
|
||||
hasher.update(encoded);
|
||||
rustfs_utils::crypto::hex(hasher.finalize().as_slice())
|
||||
}
|
||||
|
||||
pub fn ilm_recovery_export_local_process_epoch() -> Uuid {
|
||||
*ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH
|
||||
}
|
||||
|
||||
fn ilm_recovery_export_fleet_proof_matches_at(
|
||||
state: &FleetCapabilityProofState,
|
||||
proof: &IlmRecoveryExportFleetProofToken,
|
||||
expected_topology: &str,
|
||||
now: Instant,
|
||||
) -> bool {
|
||||
proof._permit.generation.is_accepting()
|
||||
&& fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
|
||||
&& state
|
||||
.proof
|
||||
.as_ref()
|
||||
.is_some_and(|current| Arc::ptr_eq(¤t.generation, &proof._permit.generation))
|
||||
}
|
||||
|
||||
fn ilm_recovery_export_fleet_proof_matches_observation_at(
|
||||
state: &FleetCapabilityProofState,
|
||||
proof: &IlmRecoveryExportFleetProofToken,
|
||||
expected_topology: &str,
|
||||
observed: &BTreeMap<String, Uuid>,
|
||||
now: Instant,
|
||||
) -> bool {
|
||||
ilm_recovery_export_fleet_proof_matches_at(state, proof, expected_topology, now)
|
||||
&& proof.token.peer_epochs.as_ref() == observed
|
||||
}
|
||||
|
||||
async fn observe_ilm_recovery_export_fleet(expected_topology: &str) -> Option<BTreeMap<String, Uuid>> {
|
||||
#[cfg(test)]
|
||||
{
|
||||
let state = ilm_recovery_export_fleet_proof_slot()
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if fleet_capability_proof_valid_at(state.proof.as_ref(), expected_topology, Instant::now()) {
|
||||
return state.proof.as_ref().map(|proof| proof.peer_epochs.as_ref().clone());
|
||||
}
|
||||
}
|
||||
let notification_sys = get_global_notification_sys()?;
|
||||
timeout(
|
||||
REMOTE_VERSION_STATE_PROBE_TIMEOUT,
|
||||
notification_sys.probe_ilm_recovery_export_fleet(expected_topology),
|
||||
)
|
||||
.await
|
||||
.ok()?
|
||||
.ok()
|
||||
}
|
||||
|
||||
async fn legacy_transition_state_reconcile_fleet_proof_matches_with_observer<F, Fut>(
|
||||
slot: &std::sync::RwLock<FleetCapabilityProofState>,
|
||||
proof: &LegacyTransitionStateReconcileFleetProofToken,
|
||||
@@ -784,7 +661,7 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() {
|
||||
state.proof.clone()
|
||||
} else {
|
||||
Some(FleetCapabilityProof::new(
|
||||
topology.clone(),
|
||||
topology,
|
||||
Arc::new(BTreeMap::new()),
|
||||
now + Duration::from_secs(60 * 60),
|
||||
))
|
||||
@@ -817,21 +694,6 @@ 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)]
|
||||
@@ -1088,7 +950,6 @@ 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);
|
||||
}
|
||||
@@ -1098,42 +959,29 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let notification_sys = get_global_notification_sys();
|
||||
let remote_version_state_probe = async {
|
||||
match notification_sys.as_ref() {
|
||||
Some(notification_sys) => timeout(
|
||||
let result = match get_global_notification_sys() {
|
||||
Some(notification_sys) => {
|
||||
match timeout(
|
||||
REMOTE_VERSION_STATE_PROBE_TIMEOUT,
|
||||
notification_sys.probe_remote_version_state_fleet(&topology_fingerprint),
|
||||
)
|
||||
.await
|
||||
.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")),
|
||||
{
|
||||
Ok(result) => result,
|
||||
Err(_) => Err(Error::other("remote version state fleet capability probe timed out")),
|
||||
}
|
||||
}
|
||||
None => Err(Error::other("remote version state 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 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 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) => {
|
||||
@@ -1156,7 +1004,6 @@ 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,
|
||||
@@ -1183,24 +1030,6 @@ 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(),
|
||||
@@ -1345,46 +1174,6 @@ impl NotificationSys {
|
||||
}
|
||||
Ok((peer_epochs, minimum_version))
|
||||
}
|
||||
|
||||
async fn probe_ilm_recovery_export_fleet(&self, topology_fingerprint: &str) -> Result<BTreeMap<String, Uuid>> {
|
||||
if self.peer_clients.len() != self.peer_topology_hosts.len() {
|
||||
return Err(Error::other("ILM recovery export capability fleet membership is incomplete"));
|
||||
}
|
||||
let local_member = runtime_sources::local_node_name().await;
|
||||
if local_member.trim().is_empty() {
|
||||
return Err(Error::other("ILM recovery export local member identity is unavailable"));
|
||||
}
|
||||
let mut peer_epochs = BTreeMap::new();
|
||||
insert_remote_version_state_peer(&mut peer_epochs, local_member.clone(), ilm_recovery_export_local_process_epoch())?;
|
||||
let probes = self.peer_clients.iter().map(|client| async {
|
||||
let client = client
|
||||
.as_ref()
|
||||
.ok_or_else(|| Error::other("ILM recovery export capability peer is unreachable"))?;
|
||||
client.probe_ilm_recovery_export(topology_fingerprint.to_string()).await
|
||||
});
|
||||
for result in join_all(probes).await {
|
||||
let (peer, epoch) = result?;
|
||||
insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
|
||||
}
|
||||
validate_ilm_recovery_export_members(&self.peer_topology_hosts, &local_member, &peer_epochs)?;
|
||||
Ok(peer_epochs)
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_ilm_recovery_export_members(
|
||||
expected_remote_members: &[String],
|
||||
local_member: &str,
|
||||
observed: &BTreeMap<String, Uuid>,
|
||||
) -> Result<()> {
|
||||
let expected = expected_remote_members
|
||||
.iter()
|
||||
.cloned()
|
||||
.chain(std::iter::once(local_member.to_string()))
|
||||
.collect::<BTreeSet<_>>();
|
||||
if expected.len() != expected_remote_members.len().saturating_add(1) || observed.keys().ne(expected.iter()) {
|
||||
return Err(Error::other("ILM recovery export capability fleet membership does not match topology"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Rolling tier activity summed over every cluster member that answered, with
|
||||
@@ -3718,81 +3507,6 @@ 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";
|
||||
|
||||
@@ -830,10 +830,6 @@ 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,
|
||||
@@ -16874,43 +16870,6 @@ 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],
|
||||
|
||||
@@ -175,7 +175,6 @@ 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;
|
||||
@@ -220,18 +219,6 @@ 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<u8> {
|
||||
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],
|
||||
@@ -2140,13 +2127,12 @@ 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,
|
||||
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,
|
||||
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,
|
||||
};
|
||||
use crate::heal_control;
|
||||
use std::time::Duration;
|
||||
@@ -2210,19 +2196,6 @@ 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 =
|
||||
|
||||
@@ -14,19 +14,18 @@
|
||||
|
||||
use crate::admin::auth::authorize_admin_request;
|
||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||
use crate::admin::runtime_sources::{current_action_credentials, object_store_from_extensions};
|
||||
use crate::admin::runtime_sources::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, 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,
|
||||
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,
|
||||
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,
|
||||
@@ -35,30 +34,21 @@ 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 aes_gcm::{
|
||||
Aes256Gcm, Key, Nonce,
|
||||
aead::{Aead, KeyInit},
|
||||
};
|
||||
use http::{HeaderMap, HeaderValue, header};
|
||||
use http::HeaderMap;
|
||||
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, base64_decode_url_safe_no_pad, base64_encode_url_safe_no_pad,
|
||||
crypto::hex_sha256,
|
||||
MaskedAccessKey,
|
||||
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;
|
||||
@@ -70,8 +60,6 @@ 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<Mutex<Vec<ManualTransitionRunScope>>> = OnceLock::new();
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
@@ -254,16 +242,6 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> 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(())
|
||||
}
|
||||
|
||||
@@ -481,14 +459,6 @@ async fn authorize_transition_admin_request(req: &S3Request<Body>, action: Admin
|
||||
Ok(actor)
|
||||
}
|
||||
|
||||
async fn authorize_recovery_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
|
||||
if req.credentials.is_none() {
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"));
|
||||
}
|
||||
let credentials = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
|
||||
Ok(recovery_actor_sha256(&credentials))
|
||||
}
|
||||
|
||||
fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uuid> {
|
||||
Uuid::parse_str(params.get("transaction_id").unwrap_or(""))
|
||||
.map_err(|_| s3_error!(InvalidArgument, "invalid transition transaction id"))
|
||||
@@ -496,19 +466,14 @@ fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uu
|
||||
|
||||
fn recovery_control_id_from_params(params: &Params<'_, '_>) -> S3Result<String> {
|
||||
let control_id = params.get("control_id").unwrap_or("");
|
||||
validate_recovery_sha256(control_id, "invalid ILM recovery control id")?;
|
||||
Ok(control_id.to_string())
|
||||
}
|
||||
|
||||
fn validate_recovery_sha256(value: &str, message: &'static str) -> S3Result<()> {
|
||||
if value.len() != 64
|
||||
|| !value
|
||||
if control_id.len() != 64
|
||||
|| !control_id
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
|
||||
{
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, message));
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control id"));
|
||||
}
|
||||
Ok(())
|
||||
Ok(control_id.to_string())
|
||||
}
|
||||
|
||||
fn map_recovery_control_error(err: StorageError) -> S3Error {
|
||||
@@ -519,394 +484,6 @@ fn map_recovery_control_error(err: StorageError) -> S3Error {
|
||||
}
|
||||
}
|
||||
|
||||
fn recovery_export_id_from_params(params: &Params<'_, '_>) -> S3Result<String> {
|
||||
let export_id = params.get("export_id").unwrap_or("");
|
||||
validate_recovery_sha256(export_id, "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<HeaderMap> {
|
||||
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, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryReceiptAction {
|
||||
Export,
|
||||
AbandonRemoteCleanup,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryReceiptMode {
|
||||
DryRun,
|
||||
Execute,
|
||||
}
|
||||
|
||||
const fn default_recovery_receipt_mode() -> IlmRecoveryReceiptMode {
|
||||
IlmRecoveryReceiptMode::Execute
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct IlmRecoveryObservationReceipt {
|
||||
schema: String,
|
||||
action: IlmRecoveryReceiptAction,
|
||||
// Receipts issued before action modes were introduced represented the
|
||||
// existing export execution path, so decode them as Execute until their
|
||||
// fixed 15-minute lifetime elapses.
|
||||
#[serde(default = "default_recovery_receipt_mode")]
|
||||
mode: IlmRecoveryReceiptMode,
|
||||
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<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
observation_receipt_expires_at_unix_nanos: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryDispositionReasonCode {
|
||||
LegacyRemoteCleanupAbandoned,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
|
||||
enum IlmRecoveryRecordMutationRequest {
|
||||
Export {
|
||||
observation_receipt: String,
|
||||
},
|
||||
AbandonRemoteCleanup {
|
||||
mode: IlmRecoveryReceiptMode,
|
||||
observation_receipt: String,
|
||||
export_id: String,
|
||||
export_sha256: String,
|
||||
reason_code: IlmRecoveryDispositionReasonCode,
|
||||
#[serde(default)]
|
||||
confirm: Option<bool>,
|
||||
#[serde(default)]
|
||||
acknowledge_remote_cleanup_abandoned: Option<bool>,
|
||||
},
|
||||
}
|
||||
|
||||
fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result<IlmRecoveryRecordMutationRequest> {
|
||||
let value: serde_json::Value = serde_json::from_slice(body)
|
||||
.map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery request"))?;
|
||||
let request: IlmRecoveryRecordMutationRequest = serde_json::from_value(value.clone())
|
||||
.map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery request"))?;
|
||||
if matches!(
|
||||
&request,
|
||||
IlmRecoveryRecordMutationRequest::AbandonRemoteCleanup {
|
||||
mode: IlmRecoveryReceiptMode::DryRun,
|
||||
..
|
||||
}
|
||||
) && value
|
||||
.as_object()
|
||||
.is_some_and(|object| object.contains_key("confirm") || object.contains_key("acknowledge_remote_cleanup_abandoned"))
|
||||
{
|
||||
return Err(admin_s3_error(
|
||||
AdminS3ErrorCode::InvalidArgument,
|
||||
"ILM recovery dry-run must not include terminal confirmation fields",
|
||||
));
|
||||
}
|
||||
Ok(request)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
enum ValidatedIlmRecoveryRecordMutation<'a> {
|
||||
Export {
|
||||
observation_receipt: &'a str,
|
||||
},
|
||||
AbandonDryRun {
|
||||
observation_receipt: &'a str,
|
||||
export_id: &'a str,
|
||||
export_sha256: &'a str,
|
||||
reason_code: IlmRecoveryDispositionReasonCode,
|
||||
},
|
||||
AbandonExecute {
|
||||
observation_receipt: &'a str,
|
||||
export_id: &'a str,
|
||||
export_sha256: &'a str,
|
||||
reason_code: IlmRecoveryDispositionReasonCode,
|
||||
},
|
||||
}
|
||||
|
||||
fn validate_recovery_record_mutation_request(
|
||||
request: &IlmRecoveryRecordMutationRequest,
|
||||
) -> S3Result<ValidatedIlmRecoveryRecordMutation<'_>> {
|
||||
match request {
|
||||
IlmRecoveryRecordMutationRequest::Export { observation_receipt } => {
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::Export { observation_receipt })
|
||||
}
|
||||
IlmRecoveryRecordMutationRequest::AbandonRemoteCleanup {
|
||||
mode,
|
||||
observation_receipt,
|
||||
export_id,
|
||||
export_sha256,
|
||||
reason_code,
|
||||
confirm,
|
||||
acknowledge_remote_cleanup_abandoned,
|
||||
} => {
|
||||
validate_recovery_sha256(export_id, "invalid ILM recovery export id")?;
|
||||
validate_recovery_sha256(export_sha256, "invalid ILM recovery export checksum")?;
|
||||
if observation_receipt.is_empty() {
|
||||
return Err(admin_s3_error(
|
||||
AdminS3ErrorCode::InvalidArgument,
|
||||
"ILM recovery observation receipt must not be empty",
|
||||
));
|
||||
}
|
||||
match mode {
|
||||
IlmRecoveryReceiptMode::DryRun if confirm.is_none() && acknowledge_remote_cleanup_abandoned.is_none() => {
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::AbandonDryRun {
|
||||
observation_receipt,
|
||||
export_id,
|
||||
export_sha256,
|
||||
reason_code: *reason_code,
|
||||
})
|
||||
}
|
||||
IlmRecoveryReceiptMode::DryRun => Err(admin_s3_error(
|
||||
AdminS3ErrorCode::InvalidArgument,
|
||||
"ILM recovery dry-run must not include terminal confirmation fields",
|
||||
)),
|
||||
IlmRecoveryReceiptMode::Execute
|
||||
if *confirm == Some(true) && *acknowledge_remote_cleanup_abandoned == Some(true) =>
|
||||
{
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::AbandonExecute {
|
||||
observation_receipt,
|
||||
export_id,
|
||||
export_sha256,
|
||||
reason_code: *reason_code,
|
||||
})
|
||||
}
|
||||
IlmRecoveryReceiptMode::Execute => Err(admin_s3_error(
|
||||
AdminS3ErrorCode::InvalidRequest,
|
||||
"ILM recovery disposition requires confirm=true and acknowledge_remote_cleanup_abandoned=true",
|
||||
)),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
struct IlmRecoveryExportCreateResponse {
|
||||
export_id: String,
|
||||
export_sha256: String,
|
||||
download_url: String,
|
||||
outcome: &'static str,
|
||||
}
|
||||
|
||||
// These response envelopes pin the future disposition wire contract before
|
||||
// its storage state machine is connected to this handler.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryDispositionDryRunStatus {
|
||||
Ready,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryDispositionState {
|
||||
Applying,
|
||||
Completed,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum IlmRecoveryDispositionOutcome {
|
||||
AcceptedForRecovery,
|
||||
Completed,
|
||||
Replayed,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct IlmRecoveryDispositionDryRunResponse {
|
||||
action: IlmRecoveryReceiptAction,
|
||||
mode: IlmRecoveryReceiptMode,
|
||||
status: IlmRecoveryDispositionDryRunStatus,
|
||||
disposition_id: String,
|
||||
export_id: String,
|
||||
export_sha256: String,
|
||||
source_generation_sha256: String,
|
||||
copy_set_sha256: String,
|
||||
source_copy_count: usize,
|
||||
observation_receipt: String,
|
||||
observation_receipt_expires_at_unix_nanos: i64,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct IlmRecoveryDispositionExecuteResponse {
|
||||
action: IlmRecoveryReceiptAction,
|
||||
mode: IlmRecoveryReceiptMode,
|
||||
disposition_id: String,
|
||||
state: IlmRecoveryDispositionState,
|
||||
outcome: IlmRecoveryDispositionOutcome,
|
||||
confirmed_absent_copy_count: usize,
|
||||
source_copy_count: usize,
|
||||
}
|
||||
|
||||
fn recovery_actor_sha256(credentials: &Credentials) -> String {
|
||||
let access_key = &credentials.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());
|
||||
hex_sha256(&bound, ToOwned::to_owned)
|
||||
}
|
||||
|
||||
fn recovery_receipt_credentials() -> S3Result<Credentials> {
|
||||
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<String> {
|
||||
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::<Aes256Gcm>::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<IlmRecoveryObservationReceipt> {
|
||||
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::<Aes256Gcm>::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,
|
||||
action: IlmRecoveryReceiptAction,
|
||||
mode: IlmRecoveryReceiptMode,
|
||||
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,
|
||||
mode,
|
||||
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,
|
||||
expected_action: IlmRecoveryReceiptAction,
|
||||
expected_mode: IlmRecoveryReceiptMode,
|
||||
now_unix_nanos: i64,
|
||||
) -> S3Result<IlmRecoveryExportObservation> {
|
||||
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 != expected_action
|
||||
|| receipt.mode != expected_mode
|
||||
|| 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"),
|
||||
@@ -1537,100 +1114,15 @@ pub struct IlmRecoveryControlInspectHandler {}
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for IlmRecoveryControlInspectHandler {
|
||||
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
let actor_sha256 = authorize_recovery_admin_request(&req, AdminAction::ListTierAction).await?;
|
||||
authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
|
||||
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.clone(), &control_id)
|
||||
let control = inspect_recovery_control(store, &control_id)
|
||||
.await
|
||||
.map_err(map_recovery_control_error)?;
|
||||
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,
|
||||
IlmRecoveryReceiptAction::Export,
|
||||
IlmRecoveryReceiptMode::Execute,
|
||||
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
let actor_sha256 = authorize_recovery_admin_request(&req, AdminAction::SetTierAction).await?;
|
||||
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 = parse_recovery_record_mutation_request(&body)?;
|
||||
let ValidatedIlmRecoveryRecordMutation::Export { observation_receipt } =
|
||||
validate_recovery_record_mutation_request(&request)?
|
||||
else {
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "unsupported ILM recovery action"));
|
||||
};
|
||||
let receipt = decode_recovery_receipt(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,
|
||||
IlmRecoveryReceiptAction::Export,
|
||||
IlmRecoveryReceiptMode::Execute,
|
||||
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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))
|
||||
json_response(StatusCode::OK, &control)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1750,350 +1242,6 @@ 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: IlmRecoveryReceiptAction::Export,
|
||||
mode: IlmRecoveryReceiptMode::Execute,
|
||||
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,
|
||||
IlmRecoveryReceiptAction::Export,
|
||||
IlmRecoveryReceiptMode::Execute,
|
||||
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,
|
||||
IlmRecoveryReceiptAction::Export,
|
||||
IlmRecoveryReceiptMode::Execute,
|
||||
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 = IlmRecoveryReceiptAction::AbandonRemoteCleanup;
|
||||
assert_denied(
|
||||
invalid,
|
||||
&payload.actor_sha256,
|
||||
&payload.observation.control_id,
|
||||
payload.issued_at_unix_nanos,
|
||||
);
|
||||
let mut invalid = payload.clone();
|
||||
invalid.mode = IlmRecoveryReceiptMode::DryRun;
|
||||
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);
|
||||
|
||||
let mut legacy_payload = serde_json::to_value(&payload).unwrap();
|
||||
legacy_payload.as_object_mut().unwrap().remove("mode");
|
||||
let legacy_payload: IlmRecoveryObservationReceipt = serde_json::from_value(legacy_payload).unwrap();
|
||||
assert_eq!(legacy_payload.mode, IlmRecoveryReceiptMode::Execute);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_actor_binding_uses_the_authenticated_presented_access_key() {
|
||||
let first = Credentials {
|
||||
access_key: "operator-a".to_string(),
|
||||
secret_key: "first-secret".to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
let same_actor_rotated_secret = Credentials {
|
||||
access_key: first.access_key.clone(),
|
||||
secret_key: "rotated-secret".to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
let other = Credentials {
|
||||
access_key: "operator-b".to_string(),
|
||||
secret_key: first.secret_key.clone(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let actor = recovery_actor_sha256(&first);
|
||||
assert_eq!(actor, recovery_actor_sha256(&same_actor_rotated_secret));
|
||||
assert_ne!(actor, recovery_actor_sha256(&other));
|
||||
assert!(!actor.contains(&first.access_key));
|
||||
|
||||
let production = include_str!("ilm_transition.rs")
|
||||
.split("\n#[cfg(test)]\n")
|
||||
.next()
|
||||
.expect("production source must precede tests");
|
||||
let gate = extract_block_between_markers(
|
||||
production,
|
||||
"async fn authorize_recovery_admin_request",
|
||||
"fn transition_transaction_id_from_params",
|
||||
);
|
||||
assert!(gate.contains("let credentials = authorize_admin_request("));
|
||||
assert!(gate.contains("recovery_actor_sha256(&credentials)"));
|
||||
assert!(!gate.contains("MaskedAccessKey"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_record_mutation_wire_contract_is_strict_and_mode_specific() {
|
||||
let export = parse_recovery_record_mutation_request(br#"{"action":"export","observation_receipt":"opaque"}"#).unwrap();
|
||||
assert!(matches!(
|
||||
validate_recovery_record_mutation_request(&export),
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::Export {
|
||||
observation_receipt: "opaque"
|
||||
})
|
||||
));
|
||||
|
||||
let export_id = "ab".repeat(32);
|
||||
let export_sha256 = "cd".repeat(32);
|
||||
let dry_run_json = format!(
|
||||
r#"{{"action":"abandon_remote_cleanup","mode":"dry_run","observation_receipt":"opaque-dry-run","export_id":"{export_id}","export_sha256":"{export_sha256}","reason_code":"legacy_remote_cleanup_abandoned"}}"#
|
||||
);
|
||||
let dry_run = parse_recovery_record_mutation_request(dry_run_json.as_bytes()).unwrap();
|
||||
assert!(matches!(
|
||||
validate_recovery_record_mutation_request(&dry_run),
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::AbandonDryRun {
|
||||
observation_receipt: "opaque-dry-run",
|
||||
export_id: observed_export_id,
|
||||
export_sha256: observed_export_sha256,
|
||||
reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned,
|
||||
}) if observed_export_id == export_id && observed_export_sha256 == export_sha256
|
||||
));
|
||||
|
||||
let execute_json = format!(
|
||||
r#"{{"action":"abandon_remote_cleanup","mode":"execute","confirm":true,"acknowledge_remote_cleanup_abandoned":true,"observation_receipt":"opaque-execute","export_id":"{export_id}","export_sha256":"{export_sha256}","reason_code":"legacy_remote_cleanup_abandoned"}}"#
|
||||
);
|
||||
let execute = parse_recovery_record_mutation_request(execute_json.as_bytes()).unwrap();
|
||||
assert!(matches!(
|
||||
validate_recovery_record_mutation_request(&execute),
|
||||
Ok(ValidatedIlmRecoveryRecordMutation::AbandonExecute {
|
||||
observation_receipt: "opaque-execute",
|
||||
export_id: observed_export_id,
|
||||
export_sha256: observed_export_sha256,
|
||||
reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned,
|
||||
}) if observed_export_id == export_id && observed_export_sha256 == export_sha256
|
||||
));
|
||||
|
||||
let dry_run_with_confirmation = dry_run_json.replace(r#""mode":"dry_run""#, r#""mode":"dry_run","confirm":false"#);
|
||||
assert!(parse_recovery_record_mutation_request(dry_run_with_confirmation.as_bytes()).is_err());
|
||||
assert!(parse_recovery_record_mutation_request(dry_run_json.replace('}', r#","confirm":null}"#).as_bytes()).is_err());
|
||||
|
||||
let uppercase_export_id = "AB".repeat(32);
|
||||
for invalid in [
|
||||
execute_json.replace(r#""confirm":true,"#, ""),
|
||||
execute_json.replace(r#""confirm":true"#, r#""confirm":false"#),
|
||||
execute_json.replace(
|
||||
r#""acknowledge_remote_cleanup_abandoned":true"#,
|
||||
r#""acknowledge_remote_cleanup_abandoned":false"#,
|
||||
),
|
||||
execute_json.replace(export_id.as_str(), uppercase_export_id.as_str()),
|
||||
execute_json.replace(export_sha256.as_str(), "too-short"),
|
||||
execute_json.replace("opaque-execute", ""),
|
||||
] {
|
||||
if let Ok(request) = parse_recovery_record_mutation_request(invalid.as_bytes()) {
|
||||
assert!(
|
||||
validate_recovery_record_mutation_request(&request).is_err(),
|
||||
"request should fail closed: {invalid}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
for invalid in [
|
||||
br#"{"action":"export","observation_receipt":"opaque","extra":true}"#.as_slice(),
|
||||
br#"{"action":"abandon_remote_cleanup","mode":"preview"}"#.as_slice(),
|
||||
br#"{"action":"unknown","observation_receipt":"opaque"}"#.as_slice(),
|
||||
] {
|
||||
assert!(parse_recovery_record_mutation_request(invalid).is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_disposition_response_wire_contract_is_closed() {
|
||||
let export = IlmRecoveryExportCreateResponse {
|
||||
export_id: "ab".repeat(32),
|
||||
export_sha256: "cd".repeat(32),
|
||||
download_url: "/rustfs/admin/v3/ilm/recovery/exports/export-id".to_string(),
|
||||
outcome: "created",
|
||||
};
|
||||
assert_eq!(
|
||||
serde_json::to_value(&export).unwrap(),
|
||||
serde_json::json!({
|
||||
"export_id": "ab".repeat(32),
|
||||
"export_sha256": "cd".repeat(32),
|
||||
"download_url": "/rustfs/admin/v3/ilm/recovery/exports/export-id",
|
||||
"outcome": "created",
|
||||
})
|
||||
);
|
||||
|
||||
let dry_run = IlmRecoveryDispositionDryRunResponse {
|
||||
action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
|
||||
mode: IlmRecoveryReceiptMode::DryRun,
|
||||
status: IlmRecoveryDispositionDryRunStatus::Ready,
|
||||
disposition_id: "ab".repeat(32),
|
||||
export_id: "cd".repeat(32),
|
||||
export_sha256: "ef".repeat(32),
|
||||
source_generation_sha256: "12".repeat(32),
|
||||
copy_set_sha256: "34".repeat(32),
|
||||
source_copy_count: 2,
|
||||
observation_receipt: "opaque-execute".to_string(),
|
||||
observation_receipt_expires_at_unix_nanos: 900_000_000_001,
|
||||
};
|
||||
let dry_run_json = serde_json::to_value(&dry_run).unwrap();
|
||||
assert_eq!(dry_run_json["action"], "abandon_remote_cleanup");
|
||||
assert_eq!(dry_run_json["mode"], "dry_run");
|
||||
assert_eq!(dry_run_json["status"], "ready");
|
||||
assert_eq!(
|
||||
serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(dry_run_json).unwrap(),
|
||||
dry_run
|
||||
);
|
||||
|
||||
let execute = IlmRecoveryDispositionExecuteResponse {
|
||||
action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
|
||||
mode: IlmRecoveryReceiptMode::Execute,
|
||||
disposition_id: "ab".repeat(32),
|
||||
state: IlmRecoveryDispositionState::Applying,
|
||||
outcome: IlmRecoveryDispositionOutcome::AcceptedForRecovery,
|
||||
confirmed_absent_copy_count: 1,
|
||||
source_copy_count: 2,
|
||||
};
|
||||
let execute_json = serde_json::to_value(&execute).unwrap();
|
||||
assert_eq!(execute_json["state"], "applying");
|
||||
assert_eq!(execute_json["outcome"], "accepted_for_recovery");
|
||||
assert_eq!(
|
||||
serde_json::from_value::<IlmRecoveryDispositionExecuteResponse>(execute_json).unwrap(),
|
||||
execute
|
||||
);
|
||||
|
||||
let mut unknown = serde_json::to_value(&dry_run).unwrap();
|
||||
unknown["unexpected"] = serde_json::json!(true);
|
||||
assert!(serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(unknown).is_err());
|
||||
}
|
||||
|
||||
#[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<Body> {
|
||||
S3Request {
|
||||
input: Body::empty(),
|
||||
@@ -2480,19 +1628,6 @@ mod tests {
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovery_admin_gate_keeps_its_missing_credentials_response() {
|
||||
let err = authorize_recovery_admin_request(
|
||||
&manual_transition_job_request(Method::GET, "/rustfs/admin/v3/ilm/recovery/controls/control-123"),
|
||||
AdminAction::ListTierAction,
|
||||
)
|
||||
.await
|
||||
.expect_err("a recovery admin request without credentials must fail");
|
||||
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transition_admin_gate_routes_through_the_shared_gate() {
|
||||
let production = include_str!("ilm_transition.rs")
|
||||
|
||||
@@ -63,6 +63,7 @@ const EVENT_ADMIN_REQUEST_STATE: &str = "admin_request_state";
|
||||
const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected";
|
||||
const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed";
|
||||
const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted";
|
||||
const POOL_ACTIVATION_FLEET_PROOF_REQUIRED: &str = "pool activation requires a live fleet capability proof";
|
||||
|
||||
fn admin_request_id(headers: &HeaderMap) -> Option<&str> {
|
||||
headers
|
||||
@@ -321,6 +322,17 @@ fn contextualize_admin_pool_api_error(
|
||||
}
|
||||
}
|
||||
|
||||
fn decommission_start_api_error(err: crate::storage_api::error::StorageError) -> ApiError {
|
||||
if crate::storage_api::capacity::is_pool_activation_fleet_proof_error(&err) {
|
||||
return ApiError {
|
||||
code: S3ErrorCode::InternalError,
|
||||
message: POOL_ACTIVATION_FLEET_PROOF_REQUIRED.to_string(),
|
||||
source: Some(Box::new(err)),
|
||||
};
|
||||
}
|
||||
ApiError::from(err)
|
||||
}
|
||||
|
||||
fn decommission_admin_not_initialized_error_with_audit(operation: &str, audit: PoolAuditContext<'_>) -> S3Error {
|
||||
error!(
|
||||
event = EVENT_ADMIN_REQUEST_FAILED,
|
||||
@@ -790,7 +802,24 @@ impl Operation for StartDecommission {
|
||||
store
|
||||
.decommission(ctx.clone(), pools_indices.clone())
|
||||
.await
|
||||
.map_err(ApiError::from)
|
||||
.map_err(|err| {
|
||||
error!(
|
||||
event = EVENT_ADMIN_REQUEST_FAILED,
|
||||
component = LOG_COMPONENT_ADMIN_API,
|
||||
subsystem = LOG_SUBSYSTEM_POOL_ADMIN,
|
||||
operation = "start_decommission",
|
||||
action = "start_decommission",
|
||||
result = "failed",
|
||||
reason = "storage_decommission_failed",
|
||||
request_id = %request_id,
|
||||
actor = %actor,
|
||||
remote_addr = %remote_addr,
|
||||
pool_indices = ?pools_indices,
|
||||
error = %err,
|
||||
"admin request failed"
|
||||
);
|
||||
decommission_start_api_error(err)
|
||||
})
|
||||
.map_err(|err| contextualize_admin_pool_api_error(err, "start decommission", &pool_context))?;
|
||||
}
|
||||
}
|
||||
@@ -1018,9 +1047,10 @@ impl Operation for ClearDecommission {
|
||||
#[cfg(test)]
|
||||
mod pools_handler_tests {
|
||||
use super::{
|
||||
AdminPoolStatus, Body, CancelDecommission, ClearDecommission, HeaderMap, ListPools, Method, Operation, Params,
|
||||
PoolAuditContext, S3ErrorCode, S3Request, StartDecommission, StatusDecommission, StatusPool, Uri,
|
||||
contextualize_admin_pool_api_error, decommission_admin_not_initialized_error_with_audit, decommission_peer_target,
|
||||
AdminPoolStatus, Body, CancelDecommission, ClearDecommission, HeaderMap, ListPools, Method, Operation,
|
||||
POOL_ACTIVATION_FLEET_PROOF_REQUIRED, Params, PoolAuditContext, S3ErrorCode, S3Request, StartDecommission,
|
||||
StatusDecommission, StatusPool, Uri, contextualize_admin_pool_api_error,
|
||||
decommission_admin_not_initialized_error_with_audit, decommission_peer_target, decommission_start_api_error,
|
||||
has_duplicate_indices, parse_mutation_pool_query, parse_pool_idx_by_id, parse_status_pool_query,
|
||||
pool_admin_missing_credentials_error, pool_admin_missing_credentials_error_with_request,
|
||||
pool_admin_pool_index_error_with_audit, pool_admin_pool_not_found_error_with_audit,
|
||||
@@ -1209,6 +1239,21 @@ mod pools_handler_tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_decommission_start_api_error_preserves_fleet_proof_retry_marker() {
|
||||
let err = crate::storage_api::error::StorageError::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED);
|
||||
|
||||
let err = decommission_start_api_error(err);
|
||||
|
||||
assert_eq!(err.code, s3s::S3ErrorCode::InternalError);
|
||||
assert_eq!(err.message, POOL_ACTIVATION_FLEET_PROOF_REQUIRED);
|
||||
assert!(err.source.is_some());
|
||||
|
||||
let unrelated = decommission_start_api_error(crate::storage_api::error::StorageError::other("disk read failed"));
|
||||
assert_eq!(unrelated.code, s3s::S3ErrorCode::InternalError);
|
||||
assert_eq!(unrelated.message, "We encountered an internal error, please try again.");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_contextualize_admin_pool_api_error_preserves_source() {
|
||||
let err = contextualize_admin_pool_api_error(
|
||||
|
||||
@@ -507,18 +507,6 @@ 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,
|
||||
@@ -2185,8 +2173,6 @@ 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);
|
||||
|
||||
@@ -215,16 +215,6 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
"/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,
|
||||
@@ -952,16 +942,6 @@ 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,
|
||||
|
||||
@@ -233,10 +233,7 @@ 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, 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,
|
||||
IlmRecoveryClassification, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls,
|
||||
};
|
||||
pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{
|
||||
TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator,
|
||||
|
||||
@@ -187,30 +187,6 @@ static NODE_CAPABILITY_SERVER_EPOCH: LazyLock<Uuid> = 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<Vec<u8>, 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<String, Arc<HealControlReplayEntry>>,
|
||||
request_id: &str,
|
||||
@@ -1027,8 +1003,7 @@ 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);
|
||||
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 {
|
||||
if remote_version_state_probe || cross_pool_fence_probe {
|
||||
let topology_member = self
|
||||
.endpoint_pools()
|
||||
.await
|
||||
@@ -1038,7 +1013,17 @@ 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 = encode_heal_capability_response(&topology_member, remote_version_state_probe, recovery_export_probe)?;
|
||||
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 canonical_response = rustfs_protos::canonical_heal_control_response_body(
|
||||
request.get_ref().version,
|
||||
&request.get_ref().topology_fingerprint,
|
||||
@@ -4055,19 +4040,6 @@ 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());
|
||||
|
||||
@@ -417,7 +417,7 @@ pub(crate) mod ecstore_bucket {
|
||||
pub(crate) mod ecstore_capacity {
|
||||
pub(crate) use rustfs_ecstore::api::capacity::{
|
||||
DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
|
||||
is_reserved_or_invalid_bucket,
|
||||
is_pool_activation_fleet_proof_error, is_reserved_or_invalid_bucket,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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, ilm_recovery_export_local_process_epoch,
|
||||
new_global_notification_sys, start_remote_version_state_fleet_probe,
|
||||
cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys,
|
||||
start_remote_version_state_fleet_probe,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1182,10 +1182,6 @@ 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<ECStore>, file: &str) -> Result<Vec<u8>> {
|
||||
ecstore_config::com::read_config(api, file).await
|
||||
}
|
||||
|
||||
@@ -18,6 +18,8 @@
|
||||
use rustfs_storage_api as storage_contracts;
|
||||
|
||||
pub(crate) mod capacity {
|
||||
pub(crate) use crate::storage::storage_api::ecstore_capacity::is_pool_activation_fleet_proof_error;
|
||||
|
||||
pub(crate) mod service {
|
||||
pub(crate) use crate::storage::storage_api::{all_local_disk, disk_drive_path, disk_endpoint};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user