gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ilm): add legacy recovery disposition records (#7292)

Browse Source
This commit is contained in:
cxymds
2026-09-06 17:38:58 +08:00
committed by GitHub
parent 7c0e1f2ef0
commit c13ee6c5ba
6 changed files with 2192 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/durable_namespace.rs
+498 -3
View File
@@ -22,7 +22,7 @@ use super::{
bucket_lifecycle_ops::{
ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token,
},
manual_transition_job, recovery_control, recovery_export, tier_delete_journal, transition_transaction,
manual_transition_job, recovery_control, recovery_disposition, recovery_export, tier_delete_journal, transition_transaction,
};
use crate::error::{Error, Result};
use crate::services::tier::tier_probe_intent;
@@ -43,6 +43,7 @@ pub(crate) enum DurableIlmRecordKind {
ManualTransitionWorkerResult,
RecoveryControl,
RecoveryExport,
RecoveryDisposition,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -119,8 +120,14 @@ pub(crate) const RECOVERY_EXPORT_NAMESPACE: DurableIlmNamespace = DurableIlmName
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; 11] = [
pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 12] = [
TIER_DELETE_JOURNAL_NAMESPACE,
TIER_DELETE_JOURNAL_V6_NAMESPACE,
TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
@@ -132,6 +139,7 @@ pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 11] = [
MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
RECOVERY_CONTROL_NAMESPACE,
RECOVERY_EXPORT_NAMESPACE,
RECOVERY_DISPOSITION_NAMESPACE,
];
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -277,6 +285,20 @@ pub(crate) enum DurableIlmRecordCheckpoint {
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 {
@@ -292,7 +314,8 @@ impl DurableIlmRecordCheckpoint {
| Self::ManualTransitionTask { content_sha256 }
| Self::ManualTransitionWorkerResult { content_sha256 }
| Self::RecoveryControl { content_sha256, .. }
| Self::RecoveryExport { content_sha256, .. } => content_sha256,
| Self::RecoveryExport { content_sha256, .. }
| Self::RecoveryDisposition { content_sha256, .. } => content_sha256,
}
}
@@ -333,6 +356,38 @@ impl DurableIlmRecordCheckpoint {
{
return Err(Error::other("durable ILM tier delete journal checkpoint is invalid"));
}
if let Self::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
&& !recovery_disposition_checkpoint_is_valid(
content_sha256,
identity_sha256,
copy_manifest_sha256,
*copy_manifest_count,
*created_at_unix_nanos,
*revision,
state.clone(),
owner_fence_sha256.as_deref(),
*owner_lease_acquired_at_unix_nanos,
*owner_lease_expires_at_unix_nanos,
confirmed_absent_sha256,
*retain_until_unix_nanos,
)
{
return Err(Error::other("durable ILM recovery disposition checkpoint is invalid"));
}
}
if self == next {
if let Self::ManualTransitionJob {
@@ -611,6 +666,83 @@ 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,
};
@@ -628,6 +760,39 @@ 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 let Self::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
&& !recovery_disposition_checkpoint_is_valid(
content_sha256,
identity_sha256,
copy_manifest_sha256,
*copy_manifest_count,
*created_at_unix_nanos,
*revision,
state.clone(),
owner_fence_sha256.as_deref(),
*owner_lease_acquired_at_unix_nanos,
*owner_lease_expires_at_unix_nanos,
confirmed_absent_sha256,
*retain_until_unix_nanos,
)
{
return false;
}
}
if let Self::TierProbeIntent { state, .. } = terminal
&& !matches!(
state,
@@ -644,6 +809,11 @@ 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;
}
@@ -769,11 +939,111 @@ 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(
content_sha256: &str,
identity_sha256: &str,
copy_manifest_sha256: &str,
copy_manifest_count: usize,
created_at_unix_nanos: i64,
revision: u64,
state: recovery_disposition::IlmRecoveryDispositionState,
owner_fence_sha256: Option<&str>,
owner_lease_acquired_at_unix_nanos: Option<i64>,
owner_lease_expires_at_unix_nanos: Option<i64>,
confirmed_absent_sha256: &[String],
retain_until_unix_nanos: i64,
) -> bool {
use recovery_disposition::IlmRecoveryDispositionState::{Applying, Completed, Prepared};
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,
@@ -1386,6 +1656,34 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
},
)
}
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()))?;
@@ -1541,6 +1839,203 @@ 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 {
crates/ecstore/src/bucket/lifecycle/mod.rs
+1
View File
@@ -25,6 +25,7 @@ mod object_handlers_common;
mod object_lock_boundary;
pub use self::core as lifecycle;
pub mod recovery_control;
pub mod recovery_disposition;
pub mod recovery_export;
mod replication_sink;
pub mod rule;
crates/ecstore/src/bucket/lifecycle/recovery_disposition.rs
+1226
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/bucket/lifecycle/recovery_export.rs
+8 -5
View File
@@ -141,7 +141,7 @@ impl IlmRecoveryExport {
{
return Err(Error::other("ILM recovery export source bytes do not match the observed generation"));
}
if export_id(&self.control_id, &self.source_generation)? != self.export_id {
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(())
@@ -289,7 +289,7 @@ pub async fn create_recovery_export(
return Err(Error::PreconditionFailed);
}
let current_source_base64 = base64_simd::STANDARD.encode_to_string(current_source_bytes);
let candidate_export_id = export_id(&current.control_id, &current.source_generation)?;
let candidate_export_id = recovery_export_id(&current.control_id, &current.source_generation)?;
let object = recovery_export_record_object_name(current.protocol, &candidate_export_id)?;
match load_recovery_export_decoded(api.clone(), &candidate_export_id).await {
Ok((existing, export)) if export_matches_observation(&export, observation) => {
@@ -552,7 +552,7 @@ fn build_export_from_source(
.checked_add(EXPORT_RETENTION_NANOS)
.ok_or_else(|| Error::other("ILM recovery export retention timestamp overflow"))?;
let export = IlmRecoveryExport {
export_id: export_id(&observation.control_id, &observation.source_generation)?,
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(),
@@ -571,7 +571,7 @@ fn build_export_from_source(
Ok(export)
}
fn export_id(control_id: &str, generation: &IlmRecoverySourceGeneration) -> Result<String> {
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")?;
@@ -760,7 +760,10 @@ mod tests {
&base64_simd::STANDARD.encode_to_string(legacy_source()),
)
.expect("export should be valid");
assert_eq!(export.export_id, export_id(&observed.control_id, &observed.source_generation).unwrap());
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);
crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs
+7 -3
View File
@@ -5548,6 +5548,12 @@ 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)),
@@ -5557,9 +5563,7 @@ fn legacy_tier_delete_recovery_descriptor(entry: &Jentry) -> Option<(&'static st
}
pub(crate) fn validate_legacy_tier_delete_recovery_source(object_name: &str, source_schema: &str, data: &[u8]) -> Result<()> {
if canonical_legacy_tier_delete_journal_identity(object_name).is_none() {
return Err(Error::other("legacy tier delete journal path is not canonical"));
}
validate_legacy_tier_delete_recovery_path(object_name)?;
let persisted: PersistedTierDeleteJournalEntry =
serde_json::from_slice(data).map_err(|err| Error::other(format!("decode tier delete journal failed: {err}")))?;
persisted.validate_legacy_recovery_shape()?;
rustfs/src/admin/handlers/ilm_transition.rs
+452 -40
View File
@@ -481,6 +481,14 @@ 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(s3_error!(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"))
@@ -488,14 +496,19 @@ 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("");
if control_id.len() != 64
|| !control_id
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
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control id"));
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, message));
}
Ok(control_id.to_string())
Ok(())
}
fn map_recovery_control_error(err: StorageError) -> S3Error {
@@ -508,13 +521,7 @@ 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("");
if export_id.len() != 64
|| !export_id
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery export id"));
}
validate_recovery_sha256(export_id, "invalid ILM recovery export id")?;
Ok(export_id.to_string())
}
@@ -547,11 +554,34 @@ fn recovery_export_download_headers(export_id: &str, encoded_len: usize) -> S3Re
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: 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,
@@ -572,11 +602,127 @@ struct IlmRecoveryControlInspectResponse {
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(deny_unknown_fields)]
struct IlmRecoveryExportCreateRequest {
action: String,
observation_receipt: String,
#[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)]
@@ -587,16 +733,68 @@ struct IlmRecoveryExportCreateResponse {
outcome: &'static str,
}
fn recovery_actor_sha256(req: &S3Request<Body>) -> S3Result<String> {
let access_key = &req
.credentials
.as_ref()
.ok_or_else(|| admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"))?
.access_key;
// 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());
Ok(hex_sha256(&bound, ToOwned::to_owned))
hex_sha256(&bound, ToOwned::to_owned)
}
fn recovery_receipt_credentials() -> S3Result<Credentials> {
@@ -662,12 +860,15 @@ fn decode_recovery_receipt(token: &str, credentials: &Credentials) -> S3Result<I
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: "export".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"))?,
@@ -684,12 +885,15 @@ 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 != "export"
|| receipt.action != expected_action
|| receipt.mode != expected_mode
|| receipt.actor_sha256 != actor_sha256
|| receipt.observation.control_id != control_id
|| receipt.nonce.is_nil()
@@ -1333,8 +1537,7 @@ pub struct IlmRecoveryControlInspectHandler {}
#[async_trait::async_trait]
impl Operation for IlmRecoveryControlInspectHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
let actor_sha256 = recovery_actor_sha256(&req)?;
let actor_sha256 = authorize_recovery_admin_request(&req, AdminAction::ListTierAction).await?;
let control_id = recovery_control_id_from_params(&params)?;
let Some(store) = object_store_from_extensions(&req.extensions) else {
return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized"));
@@ -1345,7 +1548,13 @@ impl Operation for IlmRecoveryControlInspectHandler {
let now = OffsetDateTime::now_utc();
let (export_ready, export_not_ready_reason, observation_receipt, expires_at) =
match inspect_recovery_export_observation(store, &control_id).await {
Ok(observation) => match issue_recovery_observation_receipt(observation, actor_sha256, now) {
Ok(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),
},
@@ -1369,8 +1578,7 @@ 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)>> {
authorize_transition_admin_request(&req, AdminAction::SetTierAction).await?;
let actor_sha256 = recovery_actor_sha256(&req)?;
let actor_sha256 = authorize_recovery_admin_request(&req, AdminAction::SetTierAction).await?;
let control_id = recovery_control_id_from_params(&params)?;
let Some(store) = object_store_from_extensions(&req.extensions) else {
return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized"));
@@ -1378,15 +1586,23 @@ impl Operation for IlmRecoveryExportCreateHandler {
let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|_| {
admin_s3_error(AdminS3ErrorCode::InvalidRequest, "ILM recovery export body is too large or unreadable")
})?;
let request: IlmRecoveryExportCreateRequest = serde_json::from_slice(&body)
.map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery export request"))?;
if request.action != "export" {
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(&request.observation_receipt, &recovery_receipt_credentials()?)?;
};
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, now)?;
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)?;
@@ -1576,7 +1792,8 @@ mod tests {
.unwrap();
let payload = IlmRecoveryObservationReceipt {
schema: "rustfs-ilm-recovery-observation-receipt-v1".to_string(),
action: "export".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,
@@ -1597,13 +1814,22 @@ mod tests {
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, now)
.expect_err("invalid observation receipt must be denied");
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(
@@ -1629,7 +1855,15 @@ mod tests {
payload.issued_at_unix_nanos,
);
let mut invalid = payload.clone();
invalid.action = "abandon".to_string();
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,
@@ -1668,6 +1902,184 @@ mod tests {
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", ""),
] {
match parse_recovery_record_mutation_request(invalid.as_bytes()) {
Ok(request) => assert!(
validate_recovery_record_mutation_request(&request).is_err(),
"request should fail closed: {invalid}"
),
Err(_) => {}
}
}
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]