diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 2a785247c..9105fa3dd 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -22,7 +22,7 @@ use super::{ bucket_lifecycle_ops::{ ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token, }, - manual_transition_job, recovery_control, 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, + owner_lease_acquired_at_unix_nanos: Option, + owner_lease_expires_at_unix_nanos: Option, + confirmed_absent_sha256: Vec, + 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, + owner_lease_expires_at_unix_nanos: Option, + 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 { + // 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, + ) -> 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 { diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index 2d63efb5b..7197ad832 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -25,6 +25,7 @@ mod object_handlers_common; mod object_lock_boundary; pub use self::core as lifecycle; pub mod recovery_control; +pub mod recovery_disposition; pub mod recovery_export; mod replication_sink; pub mod rule; diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_disposition.rs b/crates/ecstore/src/bucket/lifecycle/recovery_disposition.rs new file mode 100644 index 000000000..375b72a68 --- /dev/null +++ b/crates/ecstore/src/bucket/lifecycle/recovery_disposition.rs @@ -0,0 +1,1226 @@ +// 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::sync::Arc; + +use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use super::config_boundary; +use super::recovery_control::{ + IlmRecoveryClassification, IlmRecoveryProtocol, IlmRecoverySourceGeneration, load_recovery_control, observe_recovery_source, + recovery_control_record_object_name, +}; +use super::recovery_export::{IlmRecoveryExport, load_recovery_export, recovery_export_id}; +use super::tier_delete_journal::{ + TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA, validate_legacy_tier_delete_recovery_path, +}; +use crate::disk::RUSTFS_META_BUCKET; +use crate::error::{Error, Result as EcstoreResult}; +use crate::object_api::{ObjectOptions, WriteCompletion}; +use crate::storage_api_contracts::{namespace::NamespaceLocking as _, object::HTTPPreconditions}; +use crate::store::ECStore; + +pub const ILM_RECOVERY_DISPOSITION_SCHEMA: &str = "rustfs-ilm-recovery-disposition-v1"; +pub const ILM_RECOVERY_DISPOSITION_PREFIX: &str = "ilm/recovery-dispositions"; +pub const MAX_ILM_RECOVERY_DISPOSITION_SIZE: usize = 16 * 1024; +const DISPOSITION_RETENTION_NANOS: i64 = 365 * 24 * 60 * 60 * 1_000_000_000; +const OWNER_FENCE_CHECKPOINT_DOMAIN: &[u8] = b"rustfs-ilm-recovery-disposition-owner-fence-v1"; + +pub type Result = std::result::Result; + +#[derive(Debug, thiserror::Error)] +pub enum IlmRecoveryDispositionError { + #[error("ILM recovery disposition is corrupt: {0}")] + Corrupt(&'static str), + #[error("ILM recovery disposition schema is unsupported: {0}")] + UnsupportedSchema(String), + #[error("ILM recovery disposition checksum mismatch")] + ChecksumMismatch, + #[error("ILM recovery disposition successor is invalid: {0}")] + InvalidSuccessor(&'static str), + #[error("ILM recovery disposition json error: {0}")] + Json(#[from] serde_json::Error), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryDispositionAction { + AbandonRemoteCleanup, +} + +impl IlmRecoveryDispositionAction { + pub const fn as_str(self) -> &'static str { + match self { + Self::AbandonRemoteCleanup => "abandon_remote_cleanup", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryDispositionReasonCode { + LegacyRemoteCleanupAbandoned, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryDispositionState { + Prepared, + Applying, + Completed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryDispositionIdentity { + pub disposition_id: String, + pub protocol: IlmRecoveryProtocol, + pub action: IlmRecoveryDispositionAction, + pub export_id: String, + pub export_content_sha256: String, + pub control_id: String, + pub control_etag: String, + pub control_revision: u64, + pub canonical_source_path: String, + pub source_generation: IlmRecoverySourceGeneration, + pub admitted_topology_generation: String, + pub admitted_member_epochs_sha256: String, + pub actor_sha256: String, + pub reason_code: IlmRecoveryDispositionReasonCode, + pub confirmed_at_unix_nanos: i64, +} + +impl IlmRecoveryDispositionIdentity { + fn validate(&self) -> Result<()> { + validate_sha256(&self.disposition_id, "disposition ID is invalid")?; + validate_sha256(&self.export_id, "export ID is invalid")?; + validate_sha256(&self.export_content_sha256, "export content checksum is invalid")?; + validate_sha256(&self.control_id, "control ID is invalid")?; + validate_sha256(&self.admitted_topology_generation, "admitted topology generation is invalid")?; + validate_sha256(&self.admitted_member_epochs_sha256, "admitted member epoch digest is invalid")?; + validate_sha256(&self.actor_sha256, "actor digest is invalid")?; + if self.protocol != IlmRecoveryProtocol::TierDeleteJournal { + return Err(IlmRecoveryDispositionError::Corrupt( + "only legacy tier-delete journals support disposition", + )); + } + if self.control_etag.trim().is_empty() || self.control_revision == 0 { + return Err(IlmRecoveryDispositionError::Corrupt("control generation is invalid")); + } + validate_legacy_tier_delete_recovery_path(&self.canonical_source_path) + .map_err(|_| IlmRecoveryDispositionError::Corrupt("legacy source path is not canonical"))?; + self.source_generation + .validate() + .map_err(|_| IlmRecoveryDispositionError::Corrupt("source generation is invalid"))?; + if !matches!( + self.source_generation.source_schema.as_str(), + TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA | TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA + ) { + return Err(IlmRecoveryDispositionError::Corrupt( + "only legacy v1/v2 journal generations support disposition", + )); + } + if 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 + }) { + return Err(IlmRecoveryDispositionError::Corrupt( + "copy manifest does not describe one exact source generation", + )); + } + if recovery_disposition_id(&self.export_id, self.action)? != self.disposition_id { + return Err(IlmRecoveryDispositionError::Corrupt("disposition ID does not match export and action")); + } + if recovery_export_id(&self.control_id, &self.source_generation) + .map_err(|_| IlmRecoveryDispositionError::Corrupt("export identity is invalid"))? + != self.export_id + { + return Err(IlmRecoveryDispositionError::Corrupt( + "export ID does not match control and source generation", + )); + } + if self.confirmed_at_unix_nanos <= 0 { + return Err(IlmRecoveryDispositionError::Corrupt("confirmation timestamp is not positive")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryDispositionOwnerLease { + pub owner_id: String, + pub owner_epoch: Uuid, + pub lease_acquired_at_unix_nanos: i64, + pub lease_expires_at_unix_nanos: i64, + pub topology_generation: String, + pub member_epochs_sha256: String, +} + +impl IlmRecoveryDispositionOwnerLease { + fn validate(&self) -> Result<()> { + if self.owner_id.trim().is_empty() || self.owner_epoch.is_nil() { + return Err(IlmRecoveryDispositionError::Corrupt("owner fence is invalid")); + } + if self.lease_acquired_at_unix_nanos <= 0 || self.lease_expires_at_unix_nanos <= self.lease_acquired_at_unix_nanos { + return Err(IlmRecoveryDispositionError::Corrupt("owner lease interval is invalid")); + } + validate_sha256(&self.topology_generation, "owner topology generation is invalid")?; + validate_sha256(&self.member_epochs_sha256, "owner member epoch digest is invalid") + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryDisposition { + pub identity: IlmRecoveryDispositionIdentity, + pub created_at_unix_nanos: i64, + pub retain_until_unix_nanos: i64, + pub revision: u64, + pub state: IlmRecoveryDispositionState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub owner: Option, + pub confirmed_absent: Vec, +} + +impl IlmRecoveryDisposition { + pub fn new(identity: IlmRecoveryDispositionIdentity, created_at_unix_nanos: i64) -> Result { + let retain_until_unix_nanos = created_at_unix_nanos + .checked_add(DISPOSITION_RETENTION_NANOS) + .ok_or(IlmRecoveryDispositionError::Corrupt("retention timestamp overflow"))?; + let disposition = Self { + identity, + created_at_unix_nanos, + retain_until_unix_nanos, + revision: 1, + state: IlmRecoveryDispositionState::Prepared, + owner: None, + confirmed_absent: Vec::new(), + }; + disposition.validate()?; + Ok(disposition) + } + + pub fn validate(&self) -> Result<()> { + self.identity.validate()?; + if self.created_at_unix_nanos <= 0 + || self.created_at_unix_nanos < self.identity.confirmed_at_unix_nanos + || self.retain_until_unix_nanos < self.created_at_unix_nanos.saturating_add(DISPOSITION_RETENTION_NANOS) + { + return Err(IlmRecoveryDispositionError::Corrupt("disposition retention is invalid")); + } + if self.revision == 0 { + return Err(IlmRecoveryDispositionError::Corrupt("revision is zero")); + } + if let Some(owner) = &self.owner { + owner.validate()?; + } + if !is_strictly_sorted_unique(&self.confirmed_absent) { + return Err(IlmRecoveryDispositionError::Corrupt( + "confirmed-absent entries are not in unique canonical order", + )); + } + let manifest_authorities: Vec<&str> = self + .identity + .source_generation + .copies + .iter() + .map(|copy| copy.authority.as_str()) + .collect(); + if self + .confirmed_absent + .iter() + .any(|authority| manifest_authorities.binary_search(&authority.as_str()).is_err()) + { + return Err(IlmRecoveryDispositionError::Corrupt( + "confirmed-absent entry is not in the immutable copy manifest", + )); + } + match self.state { + IlmRecoveryDispositionState::Prepared if !self.confirmed_absent.is_empty() => { + Err(IlmRecoveryDispositionError::Corrupt("prepared disposition carries absence progress")) + } + IlmRecoveryDispositionState::Applying if self.owner.is_none() => { + Err(IlmRecoveryDispositionError::Corrupt("applying disposition has no owner lease")) + } + IlmRecoveryDispositionState::Completed + if self.owner.is_some() || self.confirmed_absent.len() != manifest_authorities.len() => + { + Err(IlmRecoveryDispositionError::Corrupt( + "completed disposition does not cover its entire manifest", + )) + } + _ => Ok(()), + } + } + + pub fn claim(&mut self, owner: IlmRecoveryDispositionOwnerLease) -> Result<()> { + if self.state == IlmRecoveryDispositionState::Completed || self.owner.is_some() { + return Err(IlmRecoveryDispositionError::InvalidSuccessor( + "only an unowned nonterminal disposition can be claimed", + )); + } + owner.validate()?; + self.bump_revision()?; + self.owner = Some(owner); + self.validate() + } + + pub fn take_over(&mut self, owner: IlmRecoveryDispositionOwnerLease) -> Result<()> { + let current = self + .owner + .as_ref() + .ok_or(IlmRecoveryDispositionError::InvalidSuccessor("takeover requires an existing owner"))?; + if self.state == IlmRecoveryDispositionState::Completed + || owner.owner_epoch == current.owner_epoch + || owner.lease_acquired_at_unix_nanos < current.lease_expires_at_unix_nanos + { + return Err(IlmRecoveryDispositionError::InvalidSuccessor("takeover did not fence an expired owner")); + } + owner.validate()?; + self.bump_revision()?; + self.owner = Some(owner); + self.validate() + } + + pub fn begin_applying(&mut self) -> Result<()> { + if self.state != IlmRecoveryDispositionState::Prepared || self.owner.is_none() { + return Err(IlmRecoveryDispositionError::InvalidSuccessor( + "applying requires an owned prepared disposition", + )); + } + self.bump_revision()?; + self.state = IlmRecoveryDispositionState::Applying; + self.validate() + } + + pub fn confirm_absent(&mut self, authority: impl Into) -> Result<()> { + if self.state != IlmRecoveryDispositionState::Applying || self.owner.is_none() { + return Err(IlmRecoveryDispositionError::InvalidSuccessor( + "absence progress requires an owned applying disposition", + )); + } + let authority = authority.into(); + if self + .identity + .source_generation + .copies + .binary_search_by(|copy| copy.authority.cmp(&authority)) + .is_err() + { + return Err(IlmRecoveryDispositionError::InvalidSuccessor( + "absence progress does not name a manifest entry", + )); + } + match self.confirmed_absent.binary_search(&authority) { + Ok(_) => return Err(IlmRecoveryDispositionError::InvalidSuccessor("absence progress is already recorded")), + Err(index) => self.confirmed_absent.insert(index, authority), + } + self.bump_revision()?; + self.validate() + } + + pub fn complete(&mut self) -> Result<()> { + if self.state != IlmRecoveryDispositionState::Applying + || self.confirmed_absent.len() != self.identity.source_generation.copies.len() + { + return Err(IlmRecoveryDispositionError::InvalidSuccessor( + "completion requires absence proof for the entire manifest", + )); + } + self.bump_revision()?; + self.state = IlmRecoveryDispositionState::Completed; + self.owner = None; + self.validate() + } + + pub fn validate_successor(&self, next: &Self) -> Result<()> { + self.validate()?; + next.validate()?; + if self.identity != next.identity + || self.created_at_unix_nanos != next.created_at_unix_nanos + || self.retain_until_unix_nanos != next.retain_until_unix_nanos + { + return Err(IlmRecoveryDispositionError::InvalidSuccessor("immutable identity changed")); + } + if self.revision.checked_add(1) != Some(next.revision) { + return Err(IlmRecoveryDispositionError::InvalidSuccessor("revision did not advance by one")); + } + if !is_sorted_subset(&self.confirmed_absent, &next.confirmed_absent) { + return Err(IlmRecoveryDispositionError::InvalidSuccessor("confirmed-absent progress regressed")); + } + + match (self.state, next.state) { + (IlmRecoveryDispositionState::Prepared, IlmRecoveryDispositionState::Prepared) + | (IlmRecoveryDispositionState::Applying, IlmRecoveryDispositionState::Applying) => { + self.validate_same_state_successor(next) + } + (IlmRecoveryDispositionState::Prepared, IlmRecoveryDispositionState::Applying) + if self.owner.is_some() && self.owner == next.owner && self.confirmed_absent == next.confirmed_absent => + { + Ok(()) + } + (IlmRecoveryDispositionState::Applying, IlmRecoveryDispositionState::Completed) + if self.owner.is_some() + && next.owner.is_none() + && self.confirmed_absent == next.confirmed_absent + && self.confirmed_absent.len() == self.identity.source_generation.copies.len() => + { + Ok(()) + } + _ => Err(IlmRecoveryDispositionError::InvalidSuccessor("state transition is not allowed")), + } + } + + fn is_same_or_later_generation_of(&self, previous: &Self) -> bool { + if self.identity != previous.identity + || self.created_at_unix_nanos != previous.created_at_unix_nanos + || self.retain_until_unix_nanos != previous.retain_until_unix_nanos + || self.revision < previous.revision + || !is_sorted_subset(&previous.confirmed_absent, &self.confirmed_absent) + { + return false; + } + if self.revision == previous.revision { + return self == previous; + } + let Some(revision_distance) = self.revision.checked_sub(previous.revision) else { + return false; + }; + let previous_progress = previous.confirmed_absent.len(); + let current_progress = self.confirmed_absent.len(); + let remaining = self.identity.source_generation.copies.len().saturating_sub(previous_progress); + let owner_change_is_fenced = || match (&previous.owner, &self.owner) { + (Some(previous_owner), Some(current_owner)) if previous_owner != current_owner => { + current_owner.lease_acquired_at_unix_nanos >= previous_owner.lease_expires_at_unix_nanos + } + _ => true, + }; + let minimum_distance = match (previous.state, self.state) { + (IlmRecoveryDispositionState::Prepared, IlmRecoveryDispositionState::Prepared) => { + if previous.owner.is_none() && self.owner.is_some() { + 1 + } else if previous.owner.is_some() + && self.owner.is_some() + && previous.owner != self.owner + && owner_change_is_fenced() + { + 1 + } else { + return false; + } + } + (IlmRecoveryDispositionState::Prepared, IlmRecoveryDispositionState::Applying) => { + let owner_steps = if previous.owner.is_none() { + 2 + } else if previous.owner == self.owner { + 1 + } else if self.owner.is_some() && owner_change_is_fenced() { + 2 + } else { + return false; + }; + owner_steps + current_progress as u64 + } + (IlmRecoveryDispositionState::Prepared, IlmRecoveryDispositionState::Completed) => { + u64::from(previous.owner.is_none()) + 2 + self.identity.source_generation.copies.len() as u64 + } + (IlmRecoveryDispositionState::Applying, IlmRecoveryDispositionState::Applying) => { + let progress_steps = current_progress.saturating_sub(previous_progress) as u64; + if previous.owner == self.owner && progress_steps > 0 { + progress_steps + } else if previous.owner != self.owner && self.owner.is_some() && owner_change_is_fenced() { + 1 + progress_steps + } else { + return false; + } + } + (IlmRecoveryDispositionState::Applying, IlmRecoveryDispositionState::Completed) => remaining as u64 + 1, + _ => return false, + }; + revision_distance >= minimum_distance + } + + fn validate_same_state_successor(&self, next: &Self) -> Result<()> { + match (&self.owner, &next.owner) { + (None, Some(_)) if self.confirmed_absent == next.confirmed_absent => Ok(()), + (Some(current), Some(candidate)) if current == candidate => { + if self.state == IlmRecoveryDispositionState::Prepared && self.confirmed_absent != next.confirmed_absent { + Err(IlmRecoveryDispositionError::InvalidSuccessor( + "prepared disposition advanced absence progress", + )) + } else if self.state == IlmRecoveryDispositionState::Applying + && self.confirmed_absent.len().checked_add(1) == Some(next.confirmed_absent.len()) + { + Ok(()) + } else { + Err(IlmRecoveryDispositionError::InvalidSuccessor( + "absence progress did not advance by one copy", + )) + } + } + (Some(current), Some(candidate)) + if self.confirmed_absent == next.confirmed_absent + && candidate.owner_epoch != current.owner_epoch + && candidate.lease_acquired_at_unix_nanos >= current.lease_expires_at_unix_nanos => + { + Ok(()) + } + _ => Err(IlmRecoveryDispositionError::InvalidSuccessor( + "same-state successor changed owner and progress together", + )), + } + } + + pub fn encode(&self) -> Result> { + self.validate()?; + let disposition_bytes = serde_json::to_vec(self)?; + let persisted = PersistedIlmRecoveryDisposition { + schema: ILM_RECOVERY_DISPOSITION_SCHEMA.to_string(), + content_sha256: hex_sha256(&disposition_bytes, ToOwned::to_owned), + disposition: self.clone(), + }; + let encoded = serde_json::to_vec(&persisted)?; + if encoded.len() > MAX_ILM_RECOVERY_DISPOSITION_SIZE { + return Err(IlmRecoveryDispositionError::Corrupt("encoded disposition exceeds maximum size")); + } + Ok(encoded) + } + + pub fn decode(expected_disposition_id: &str, data: &[u8]) -> Result { + validate_sha256(expected_disposition_id, "disposition ID is invalid")?; + if data.len() > MAX_ILM_RECOVERY_DISPOSITION_SIZE { + return Err(IlmRecoveryDispositionError::Corrupt("encoded disposition exceeds maximum size")); + } + let persisted: PersistedIlmRecoveryDisposition = serde_json::from_slice(data)?; + if persisted.schema != ILM_RECOVERY_DISPOSITION_SCHEMA { + return Err(IlmRecoveryDispositionError::UnsupportedSchema(persisted.schema)); + } + validate_sha256(&persisted.content_sha256, "content checksum is invalid")?; + let disposition_bytes = serde_json::to_vec(&persisted.disposition)?; + if hex_sha256(&disposition_bytes, ToOwned::to_owned) != persisted.content_sha256 { + return Err(IlmRecoveryDispositionError::ChecksumMismatch); + } + persisted.disposition.validate()?; + if persisted.disposition.identity.disposition_id != expected_disposition_id { + return Err(IlmRecoveryDispositionError::Corrupt("disposition ID does not match record key")); + } + Ok(persisted.disposition) + } + + fn bump_revision(&mut self) -> Result<()> { + self.revision = self + .revision + .checked_add(1) + .ok_or(IlmRecoveryDispositionError::Corrupt("revision overflow"))?; + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedIlmRecoveryDisposition { + schema: String, + content_sha256: String, + disposition: IlmRecoveryDisposition, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObservedIlmRecoveryDisposition { + pub disposition: IlmRecoveryDisposition, + pub etag: String, + pub content_sha256: String, + pub encoded: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CreatedIlmRecoveryDisposition { + pub observed: ObservedIlmRecoveryDisposition, + pub replayed: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct DecodedIlmRecoveryDispositionCheckpoint { + pub(crate) disposition_id: String, + pub(crate) content_sha256: String, + pub(crate) identity_sha256: String, + pub(crate) copy_manifest_sha256: String, + pub(crate) copy_manifest_count: usize, + pub(crate) created_at_unix_nanos: i64, + pub(crate) revision: u64, + pub(crate) state: IlmRecoveryDispositionState, + pub(crate) owner_fence_sha256: Option, + pub(crate) owner_lease_acquired_at_unix_nanos: Option, + pub(crate) owner_lease_expires_at_unix_nanos: Option, + pub(crate) confirmed_absent_sha256: Vec, + pub(crate) retain_until_unix_nanos: i64, +} + +pub fn recovery_disposition_id(export_id: &str, action: IlmRecoveryDispositionAction) -> Result { + validate_sha256(export_id, "export ID is invalid")?; + Ok(length_delimited_digest(&[export_id.as_bytes(), action.as_str().as_bytes()])) +} + +pub fn recovery_disposition_record_object_name(protocol: IlmRecoveryProtocol, disposition_id: &str) -> Result { + validate_sha256(disposition_id, "disposition ID is invalid")?; + if protocol != IlmRecoveryProtocol::TierDeleteJournal { + return Err(IlmRecoveryDispositionError::Corrupt("disposition protocol is unsupported")); + } + Ok(format!( + "{}/{}/{}/{}/{}.json", + ILM_RECOVERY_DISPOSITION_PREFIX, + protocol.as_str(), + &disposition_id[..2], + &disposition_id[2..4], + disposition_id + )) +} + +pub fn recovery_disposition_id_from_record_object_name(object: &str) -> Result<(IlmRecoveryProtocol, String)> { + let suffix = object + .strip_prefix(ILM_RECOVERY_DISPOSITION_PREFIX) + .and_then(|suffix| suffix.strip_prefix('/')) + .ok_or(IlmRecoveryDispositionError::Corrupt("disposition record path has wrong prefix"))?; + let mut parts = suffix.split('/'); + let protocol = match parts.next() { + Some("tier_delete_journal") => IlmRecoveryProtocol::TierDeleteJournal, + _ => { + return Err(IlmRecoveryDispositionError::Corrupt("disposition record protocol is invalid")); + } + }; + let shard_a = parts + .next() + .ok_or(IlmRecoveryDispositionError::Corrupt("disposition record path is incomplete"))?; + let shard_b = parts + .next() + .ok_or(IlmRecoveryDispositionError::Corrupt("disposition record path is incomplete"))?; + let disposition_id = parts + .next() + .and_then(|name| name.strip_suffix(".json")) + .ok_or(IlmRecoveryDispositionError::Corrupt("disposition record suffix is invalid"))?; + if parts.next().is_some() { + return Err(IlmRecoveryDispositionError::Corrupt("disposition record path is not canonical")); + } + validate_sha256(disposition_id, "disposition ID is invalid")?; + if shard_a != &disposition_id[..2] || shard_b != &disposition_id[2..4] { + return Err(IlmRecoveryDispositionError::Corrupt( + "disposition record shard does not match disposition ID", + )); + } + Ok((protocol, disposition_id.to_string())) +} + +pub(crate) fn decode_recovery_disposition_checkpoint( + path: &str, + data: &[u8], +) -> EcstoreResult { + let (protocol, disposition_id) = recovery_disposition_id_from_record_object_name(path).map_err(disposition_store_error)?; + let disposition = IlmRecoveryDisposition::decode(&disposition_id, data).map_err(disposition_store_error)?; + let canonical = recovery_disposition_record_object_name(protocol, &disposition_id).map_err(disposition_store_error)?; + if canonical != path || disposition.identity.protocol != protocol { + return Err(Error::other("ILM recovery disposition path is not canonical")); + } + let identity_sha256 = checkpoint_hash(&disposition.identity)?; + let owner_fence_sha256 = disposition + .owner + .as_ref() + .map(|owner| checkpoint_domain_hash(OWNER_FENCE_CHECKPOINT_DOMAIN, owner)) + .transpose()?; + let mut confirmed_absent_sha256 = disposition + .confirmed_absent + .iter() + .map(|authority| hex_sha256(authority.as_bytes(), ToOwned::to_owned)) + .collect::>(); + confirmed_absent_sha256.sort_unstable(); + Ok(DecodedIlmRecoveryDispositionCheckpoint { + disposition_id, + content_sha256: hex_sha256(data, ToOwned::to_owned), + identity_sha256, + copy_manifest_sha256: disposition.identity.source_generation.copy_set_sha256.clone(), + copy_manifest_count: disposition.identity.source_generation.copies.len(), + created_at_unix_nanos: disposition.created_at_unix_nanos, + revision: disposition.revision, + state: disposition.state, + owner_fence_sha256, + owner_lease_acquired_at_unix_nanos: disposition.owner.as_ref().map(|owner| owner.lease_acquired_at_unix_nanos), + owner_lease_expires_at_unix_nanos: disposition.owner.as_ref().map(|owner| owner.lease_expires_at_unix_nanos), + confirmed_absent_sha256, + retain_until_unix_nanos: disposition.retain_until_unix_nanos, + }) +} + +pub async fn create_recovery_disposition_if_absent( + api: Arc, + disposition: &IlmRecoveryDisposition, +) -> EcstoreResult { + if disposition.revision != 1 + || disposition.state != IlmRecoveryDispositionState::Prepared + || disposition.owner.is_some() + || !disposition.confirmed_absent.is_empty() + { + return Err(Error::PreconditionFailed); + } + let object = recovery_disposition_record_object_name(disposition.identity.protocol, &disposition.identity.disposition_id) + .map_err(disposition_store_error)?; + match load_recovery_disposition(api.clone(), disposition.identity.protocol, &disposition.identity.disposition_id).await { + Ok(observed) if observed.disposition.is_same_or_later_generation_of(disposition) => { + record_disposition_decommission_checkpoint(api.as_ref(), &object, &observed).await?; + return Ok(CreatedIlmRecoveryDisposition { + observed, + replayed: true, + }); + } + Ok(_) => return Err(Error::PreconditionFailed), + Err(err) if disposition_is_missing(&err) => {} + Err(err) => return Err(err), + } + let control_object = recovery_control_record_object_name(disposition.identity.protocol, &disposition.identity.control_id) + .map_err(|err| Error::other(err.to_string()))?; + 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, &disposition.identity.canonical_source_path) + .await?; + let source_guard = source_lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?; + let locks_current = || !control_guard.is_lock_lost() && !source_guard.is_lock_lost(); + validate_disposition_sources(api.clone(), disposition).await?; + let current_source = observe_recovery_source( + api.clone(), + &disposition.identity.canonical_source_path, + &disposition.identity.source_generation.source_schema, + ) + .await?; + if current_source.canonical_data.is_none() + || current_source.generation != disposition.identity.source_generation + || !locks_current() + { + return Err(Error::PreconditionFailed); + } + let encoded = disposition.encode().map_err(disposition_store_error)?; + 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(&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; + match load_recovery_disposition(api.clone(), disposition.identity.protocol, &disposition.identity.disposition_id).await { + Ok(observed) if observed.disposition.is_same_or_later_generation_of(disposition) => { + record_disposition_decommission_checkpoint(api.as_ref(), &object, &observed).await?; + if !locks_current() { + return Err(Error::PreconditionFailed); + } + let replayed = write_result.is_err() || observed.encoded != encoded; + Ok(CreatedIlmRecoveryDisposition { observed, replayed }) + } + Ok(_) => Err(Error::PreconditionFailed), + Err(read_err) => Err(write_result.err().unwrap_or(read_err)), + } +} + +async fn validate_disposition_sources(api: Arc, disposition: &IlmRecoveryDisposition) -> EcstoreResult<()> { + let stored_export = load_recovery_export(api.clone(), &disposition.identity.export_id).await?; + if stored_export.content_sha256 != disposition.identity.export_content_sha256 { + return Err(Error::PreconditionFailed); + } + let export = IlmRecoveryExport::decode(&stored_export.export_id, &stored_export.encoded)?; + let observed_control = load_recovery_control(api, disposition.identity.protocol, &disposition.identity.control_id).await?; + if export.control_id != disposition.identity.control_id + || export.protocol != disposition.identity.protocol + || export.control_etag != disposition.identity.control_etag + || export.control_revision != disposition.identity.control_revision + || export.classification != IlmRecoveryClassification::RetainedAmbiguous + || export.topology_generation != disposition.identity.admitted_topology_generation + || export.member_epochs_sha256 != disposition.identity.admitted_member_epochs_sha256 + || export.canonical_source_path != disposition.identity.canonical_source_path + || export.source_generation != disposition.identity.source_generation + || observed_control.etag != disposition.identity.control_etag + || observed_control.control.revision != disposition.identity.control_revision + || observed_control.control.classification != IlmRecoveryClassification::RetainedAmbiguous + || observed_control.control.identity.canonical_source_path != disposition.identity.canonical_source_path + || observed_control.control.observed_source_generation != disposition.identity.source_generation + { + return Err(Error::PreconditionFailed); + } + Ok(()) +} + +pub async fn load_recovery_disposition( + api: Arc, + protocol: IlmRecoveryProtocol, + disposition_id: &str, +) -> EcstoreResult { + let object = recovery_disposition_record_object_name(protocol, disposition_id).map_err(disposition_store_error)?; + let (encoded, metadata) = config_boundary::read_config_limited_preserve_empty_with_metadata( + api, + &object, + &ObjectOptions::default(), + MAX_ILM_RECOVERY_DISPOSITION_SIZE, + ) + .await?; + let etag = metadata + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("ILM recovery disposition is missing an ETag"))?; + let disposition = IlmRecoveryDisposition::decode(disposition_id, &encoded).map_err(disposition_store_error)?; + if disposition.identity.protocol != protocol { + return Err(Error::other("ILM recovery disposition protocol does not match record path")); + } + Ok(ObservedIlmRecoveryDisposition { + disposition, + etag, + content_sha256: hex_sha256(&encoded, ToOwned::to_owned), + encoded, + }) +} + +pub async fn save_recovery_disposition_if_current( + api: Arc, + current: &ObservedIlmRecoveryDisposition, + next: &IlmRecoveryDisposition, +) -> EcstoreResult { + current + .disposition + .validate_successor(next) + .map_err(disposition_store_error)?; + let protocol = current.disposition.identity.protocol; + let disposition_id = ¤t.disposition.identity.disposition_id; + let object = recovery_disposition_record_object_name(protocol, disposition_id).map_err(disposition_store_error)?; + let authoritative = load_recovery_disposition(api.clone(), protocol, disposition_id).await?; + if authoritative != *current { + if authoritative.disposition.is_same_or_later_generation_of(next) { + record_disposition_decommission_checkpoint(api.as_ref(), &object, &authoritative).await?; + return Ok(authoritative); + } + return Err(Error::PreconditionFailed); + } + let encoded = next.encode().map_err(disposition_store_error)?; + let write_result = config_boundary::save_config_with_opts( + api.clone(), + &object, + encoded.clone(), + &ObjectOptions { + max_parity: true, + write_completion: WriteCompletion::TailDrained, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(current.etag.clone()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await; + match load_recovery_disposition(api.clone(), protocol, disposition_id).await { + Ok(observed) if observed.disposition.is_same_or_later_generation_of(next) => { + record_disposition_decommission_checkpoint(api.as_ref(), &object, &observed).await?; + Ok(observed) + } + Ok(_) => Err(Error::PreconditionFailed), + Err(read_err) => Err(write_result.err().unwrap_or(read_err)), + } +} + +async fn record_disposition_decommission_checkpoint( + api: &ECStore, + object: &str, + observed: &ObservedIlmRecoveryDisposition, +) -> EcstoreResult<()> { + if observed.disposition.state == IlmRecoveryDispositionState::Completed { + api.record_durable_ilm_decommission_terminal(object, &observed.encoded).await + } else { + api.record_durable_ilm_decommission_progress(object, &observed.encoded).await + } +} + +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(IlmRecoveryDispositionError::Corrupt(message)); + } + Ok(()) +} + +fn is_strictly_sorted_unique(values: &[String]) -> bool { + values.windows(2).all(|pair| pair[0] < pair[1]) +} + +fn is_sorted_subset(previous: &[String], next: &[String]) -> bool { + previous.iter().all(|value| next.binary_search(value).is_ok()) +} + +fn length_delimited_digest(parts: &[&[u8]]) -> String { + let mut encoded = Vec::new(); + for part in parts { + encoded.extend_from_slice(&(part.len() as u64).to_be_bytes()); + encoded.extend_from_slice(part); + } + hex_sha256(&encoded, ToOwned::to_owned) +} + +fn checkpoint_hash(value: &T) -> EcstoreResult { + let encoded = serde_json::to_vec(value).map_err(Error::other)?; + Ok(hex_sha256(&encoded, ToOwned::to_owned)) +} + +fn checkpoint_domain_hash(domain: &[u8], value: &T) -> EcstoreResult { + let encoded = serde_json::to_vec(value).map_err(Error::other)?; + Ok(length_delimited_digest(&[domain, &encoded])) +} + +fn disposition_store_error(err: IlmRecoveryDispositionError) -> Error { + Error::other(err) +} + +fn disposition_is_missing(err: &Error) -> bool { + matches!( + err, + Error::ConfigNotFound | Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::VersionNotFound(_, _, _) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bucket::lifecycle::recovery_control::IlmRecoverySourceCopy; + + fn digest(value: &[u8]) -> String { + hex_sha256(value, ToOwned::to_owned) + } + + fn sample_disposition() -> IlmRecoveryDisposition { + let source_path = format!("ilm/tier-delete-journal/{}.json", digest(b"journal-identity")); + let content_sha256 = digest(b"legacy-source"); + let source_generation = IlmRecoverySourceGeneration::new( + "rustfs-tier-delete-journal-v2", + "source-etag", + content_sha256.clone(), + vec![ + IlmRecoverySourceCopy { + authority: "pool-1/set-0".to_string(), + canonical_path: source_path.clone(), + etag: "source-etag".to_string(), + encoded_len: 13, + content_sha256: content_sha256.clone(), + }, + IlmRecoverySourceCopy { + authority: "pool-0/set-0".to_string(), + canonical_path: source_path.clone(), + etag: "source-etag".to_string(), + encoded_len: 13, + content_sha256, + }, + ], + ) + .expect("source generation should be valid"); + let control_id = digest(b"control"); + let export_id = recovery_export_id(&control_id, &source_generation).expect("export ID should be valid"); + let action = IlmRecoveryDispositionAction::AbandonRemoteCleanup; + let identity = IlmRecoveryDispositionIdentity { + disposition_id: recovery_disposition_id(&export_id, action).expect("ID should be valid"), + protocol: IlmRecoveryProtocol::TierDeleteJournal, + action, + export_id, + export_content_sha256: digest(b"export-envelope"), + control_id, + control_etag: "control-etag".to_string(), + control_revision: 1, + canonical_source_path: source_path, + source_generation, + admitted_topology_generation: digest(b"topology"), + admitted_member_epochs_sha256: digest(b"member-epochs"), + actor_sha256: digest(b"actor"), + reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned, + confirmed_at_unix_nanos: 1_000_000_000, + }; + IlmRecoveryDisposition::new(identity, 1_000_000_001).expect("disposition should be valid") + } + + fn owner(epoch: u128, acquired: i64) -> IlmRecoveryDispositionOwnerLease { + IlmRecoveryDispositionOwnerLease { + owner_id: "node-a".to_string(), + owner_epoch: Uuid::from_u128(epoch), + lease_acquired_at_unix_nanos: acquired, + lease_expires_at_unix_nanos: acquired + 1_000, + topology_generation: digest(b"topology-now"), + member_epochs_sha256: digest(b"member-epochs-now"), + } + } + + #[test] + fn disposition_id_and_path_are_canonical_and_deterministic() { + let disposition = sample_disposition(); + let id = &disposition.identity.disposition_id; + assert_eq!( + recovery_disposition_id(&disposition.identity.export_id, IlmRecoveryDispositionAction::AbandonRemoteCleanup) + .expect("ID should be valid"), + *id + ); + let path = + recovery_disposition_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, id).expect("path should be valid"); + assert_eq!( + recovery_disposition_id_from_record_object_name(&path).expect("path should parse"), + (IlmRecoveryProtocol::TierDeleteJournal, id.clone()) + ); + let wrong_shard = path.replacen(&format!("/{}/", &id[..2]), "/ff/", 1); + assert!(recovery_disposition_id_from_record_object_name(&wrong_shard).is_err()); + } + + #[test] + fn checksum_envelope_round_trips_and_rejects_tampering() { + let disposition = sample_disposition(); + let encoded = disposition.encode().expect("disposition should encode"); + assert_eq!( + IlmRecoveryDisposition::decode(&disposition.identity.disposition_id, &encoded).expect("disposition should decode"), + disposition + ); + + let mut envelope: serde_json::Value = serde_json::from_slice(&encoded).expect("fixture should be json"); + envelope["disposition"]["identity"]["control_etag"] = serde_json::Value::String("tampered".to_string()); + let tampered = serde_json::to_vec(&envelope).expect("fixture should encode"); + assert!(matches!( + IlmRecoveryDisposition::decode(&disposition.identity.disposition_id, &tampered), + Err(IlmRecoveryDispositionError::ChecksumMismatch) + )); + + envelope["unknown"] = serde_json::Value::Bool(true); + let unknown = serde_json::to_vec(&envelope).expect("fixture should encode"); + assert!(matches!( + IlmRecoveryDisposition::decode(&disposition.identity.disposition_id, &unknown), + Err(IlmRecoveryDispositionError::Json(_)) + )); + + for source_path in [ + "ilm/tier-delete-journal/legacy.json", + "ilm/tier-delete-journal/../legacy.json", + ] { + let mut invalid = disposition.clone(); + invalid.identity.canonical_source_path = source_path.to_string(); + for copy in &mut invalid.identity.source_generation.copies { + copy.canonical_path = source_path.to_string(); + } + invalid.identity.source_generation = IlmRecoverySourceGeneration::new( + invalid.identity.source_generation.source_schema.clone(), + invalid.identity.source_generation.source_etag.clone(), + invalid.identity.source_generation.content_sha256.clone(), + invalid.identity.source_generation.copies.clone(), + ) + .expect("copy manifest should be internally consistent"); + assert!(invalid.validate().is_err(), "noncanonical source path must fail closed"); + } + + let mut rebound = disposition.clone(); + rebound.identity.control_id = digest(b"different-control"); + assert!(rebound.validate().is_err(), "export ID must bind the exact control and source generation"); + } + + #[test] + fn state_and_absence_progress_successors_are_strict() { + let prepared = sample_disposition(); + let mut claimed = prepared.clone(); + claimed.claim(owner(1, 2_000_000_000)).expect("claim should succeed"); + prepared.validate_successor(&claimed).expect("claim should be a successor"); + + let mut applying = claimed.clone(); + applying.begin_applying().expect("begin applying should succeed"); + claimed.validate_successor(&applying).expect("applying should follow a claim"); + + let mut one_absent = applying.clone(); + one_absent + .confirm_absent("pool-0/set-0") + .expect("manifest entry should be recorded"); + applying + .validate_successor(&one_absent) + .expect("absence should advance monotonically"); + + let mut batched_absence = applying.clone(); + batched_absence.revision += 1; + batched_absence.confirmed_absent = vec!["pool-0/set-0".to_string(), "pool-1/set-0".to_string()]; + assert!( + applying.validate_successor(&batched_absence).is_err(), + "each durable progress revision must confirm exactly one copy" + ); + + let mut skipped_progress = one_absent.clone(); + skipped_progress.revision += 1; + skipped_progress.state = IlmRecoveryDispositionState::Completed; + skipped_progress.owner = None; + skipped_progress.confirmed_absent.push("pool-1/set-0".to_string()); + assert!(one_absent.validate_successor(&skipped_progress).is_err()); + + let mut all_absent = one_absent.clone(); + all_absent + .confirm_absent("pool-1/set-0") + .expect("manifest entry should be recorded"); + let mut completed = all_absent.clone(); + completed.complete().expect("full absence should complete"); + all_absent + .validate_successor(&completed) + .expect("completed should follow full absence"); + assert!(completed.validate_successor(&completed).is_err()); + + let mut regressed = all_absent.clone(); + regressed.revision += 1; + regressed.confirmed_absent.remove(0); + assert!(one_absent.validate_successor(®ressed).is_err()); + } + + #[test] + fn advanced_replay_requires_a_reachable_generation() { + let prepared = sample_disposition(); + + let mut skipped_to_applying = prepared.clone(); + skipped_to_applying.revision = 2; + skipped_to_applying.state = IlmRecoveryDispositionState::Applying; + skipped_to_applying.owner = Some(owner(1, 2_000_000_000)); + skipped_to_applying.confirmed_absent.push("pool-0/set-0".to_string()); + assert!(skipped_to_applying.validate().is_ok()); + assert!(!skipped_to_applying.is_same_or_later_generation_of(&prepared)); + + let mut skipped_to_completed = prepared.clone(); + skipped_to_completed.revision = 2; + skipped_to_completed.state = IlmRecoveryDispositionState::Completed; + skipped_to_completed.confirmed_absent = vec!["pool-0/set-0".to_string(), "pool-1/set-0".to_string()]; + assert!(skipped_to_completed.validate().is_ok()); + assert!(!skipped_to_completed.is_same_or_later_generation_of(&prepared)); + + let mut claimed = prepared.clone(); + claimed.claim(owner(1, 2_000_000_000)).unwrap(); + let mut applying = claimed.clone(); + applying.begin_applying().unwrap(); + applying.confirm_absent("pool-0/set-0").unwrap(); + applying.confirm_absent("pool-1/set-0").unwrap(); + let mut completed = applying.clone(); + completed.complete().unwrap(); + assert!(completed.is_same_or_later_generation_of(&prepared)); + + let mut same_revision_conflict = prepared.clone(); + same_revision_conflict.identity.actor_sha256 = digest(b"different-actor"); + assert!(!same_revision_conflict.is_same_or_later_generation_of(&prepared)); + } + + #[test] + fn owner_takeover_requires_expiry_and_cannot_advance_progress() { + let mut applying = sample_disposition(); + applying.claim(owner(1, 2_000_000_000)).expect("claim should succeed"); + applying.begin_applying().expect("begin applying should succeed"); + + let mut early = applying.clone(); + assert!(early.take_over(owner(2, 2_000_000_999)).is_err()); + + let mut taken_over = applying.clone(); + taken_over + .take_over(owner(2, 2_000_001_000)) + .expect("expired owner should be replaceable"); + applying + .validate_successor(&taken_over) + .expect("takeover should be a valid successor"); + + let mut changed_both = taken_over.clone(); + changed_both.revision += 1; + changed_both.owner = Some(owner(3, 2_000_002_000)); + changed_both.confirmed_absent.push("pool-0/set-0".to_string()); + assert!(taken_over.validate_successor(&changed_both).is_err()); + } + + #[test] + fn manifest_and_confirmed_absent_must_remain_canonical() { + let mut disposition = sample_disposition(); + disposition.identity.source_generation.copies.swap(0, 1); + assert!(disposition.validate().is_err()); + + let mut disposition = sample_disposition(); + disposition.state = IlmRecoveryDispositionState::Applying; + disposition.owner = Some(owner(1, 2_000_000_000)); + disposition.confirmed_absent = vec!["pool-9/set-9".to_string()]; + assert!(disposition.validate().is_err()); + + let mut disposition = sample_disposition(); + disposition.state = IlmRecoveryDispositionState::Completed; + disposition.confirmed_absent = vec!["pool-0/set-0".to_string()]; + assert!(disposition.validate().is_err()); + } + + #[test] + fn encoded_disposition_is_bounded_to_sixteen_kibibytes() { + let mut disposition = sample_disposition(); + disposition.identity.control_etag = "x".repeat(MAX_ILM_RECOVERY_DISPOSITION_SIZE); + assert!(matches!( + disposition.encode(), + Err(IlmRecoveryDispositionError::Corrupt("encoded disposition exceeds maximum size")) + )); + assert!(matches!( + IlmRecoveryDisposition::decode( + &sample_disposition().identity.disposition_id, + &vec![b'x'; MAX_ILM_RECOVERY_DISPOSITION_SIZE + 1] + ), + Err(IlmRecoveryDispositionError::Corrupt("encoded disposition exceeds maximum size")) + )); + } + + #[test] + fn durable_checkpoint_hashes_identity_owner_manifest_and_absence_set() { + let prepared = sample_disposition(); + let path = + recovery_disposition_record_object_name(IlmRecoveryProtocol::TierDeleteJournal, &prepared.identity.disposition_id) + .expect("path should be valid"); + let prepared_encoded = prepared.encode().expect("prepared disposition should encode"); + let prepared_checkpoint = + decode_recovery_disposition_checkpoint(&path, &prepared_encoded).expect("prepared checkpoint should decode"); + assert_eq!(prepared_checkpoint.state, IlmRecoveryDispositionState::Prepared); + assert_eq!(prepared_checkpoint.owner_fence_sha256, None); + + let mut disposition = prepared; + disposition.claim(owner(1, 2_000_000_000)).expect("claim should succeed"); + disposition.begin_applying().expect("begin applying should succeed"); + disposition + .confirm_absent("pool-0/set-0") + .expect("absence should be recorded"); + let encoded = disposition.encode().expect("disposition should encode"); + let checkpoint = decode_recovery_disposition_checkpoint(&path, &encoded).expect("checkpoint should decode"); + assert_eq!(checkpoint.disposition_id, disposition.identity.disposition_id); + assert_eq!(checkpoint.content_sha256, digest(&encoded)); + assert_eq!(checkpoint.copy_manifest_sha256, disposition.identity.source_generation.copy_set_sha256); + assert_eq!(checkpoint.copy_manifest_count, 2); + assert_eq!(checkpoint.revision, disposition.revision); + assert_eq!(checkpoint.state, IlmRecoveryDispositionState::Applying); + assert_eq!( + checkpoint.owner_fence_sha256, + Some( + checkpoint_domain_hash( + OWNER_FENCE_CHECKPOINT_DOMAIN, + disposition.owner.as_ref().expect("applying disposition should have an owner") + ) + .expect("owner fence should hash") + ) + ); + assert_eq!(checkpoint.confirmed_absent_sha256, vec![digest(b"pool-0/set-0")]); + assert_eq!(checkpoint.retain_until_unix_nanos, disposition.retain_until_unix_nanos); + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_export.rs b/crates/ecstore/src/bucket/lifecycle/recovery_export.rs index 52048f476..009e81a0e 100644 --- a/crates/ecstore/src/bucket/lifecycle/recovery_export.rs +++ b/crates/ecstore/src/bucket/lifecycle/recovery_export.rs @@ -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(¤t.control_id, ¤t.source_generation)?; + 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) => { @@ -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 { +pub(crate) fn recovery_export_id(control_id: &str, generation: &IlmRecoverySourceGeneration) -> Result { validate_sha256(control_id, "ILM recovery export control ID is invalid")?; validate_sha256(&generation.content_sha256, "ILM recovery export source checksum is invalid")?; validate_sha256(&generation.copy_set_sha256, "ILM recovery export copy-set checksum is invalid")?; @@ -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); diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 18613523b..ea3a30447 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -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()?; diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index 3562fecc6..d490914cf 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -481,6 +481,14 @@ async fn authorize_transition_admin_request(req: &S3Request, action: Admin Ok(actor) } +async fn authorize_recovery_admin_request(req: &S3Request, action: AdminAction) -> S3Result { + 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::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) -> S3Result { 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 { 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, } +#[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, + #[serde(default)] + acknowledge_remote_cleanup_abandoned: Option, + }, +} + +fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result { + 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> { + 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) -> S3Result { - 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 { @@ -662,12 +860,15 @@ fn decode_recovery_receipt(token: &str, credentials: &Credentials) -> S3Result 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 { 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, params: Params<'_, '_>) -> S3Result> { - 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(¶ms)?; 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, params: Params<'_, '_>) -> S3Result> { - 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(¶ms)?; 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::(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::(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::(unknown).is_err()); } #[test]