From 847d3f2433998c1bb50a12dad11dc1cf4bce53f9 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 5 Sep 2026 18:40:15 +0800 Subject: [PATCH] fix(ilm): persist bounded transition recovery controls --- crates/ecstore/src/api/mod.rs | 7 + .../src/bucket/lifecycle/durable_namespace.rs | 223 ++- crates/ecstore/src/bucket/lifecycle/mod.rs | 1 + .../src/bucket/lifecycle/recovery_control.rs | 1280 +++++++++++++++++ .../lifecycle/transition_transaction.rs | 656 ++++++++- crates/ecstore/src/store/init.rs | 305 +++- rustfs/src/admin/handlers/ilm_transition.rs | 150 +- rustfs/src/admin/storage_api.rs | 3 + 8 files changed, 2541 insertions(+), 84 deletions(-) create mode 100644 crates/ecstore/src/bucket/lifecycle/recovery_control.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index fd66c897a..34d89daec 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -69,6 +69,13 @@ pub mod bucket { }; } + pub mod recovery_control { + pub use crate::bucket::lifecycle::recovery_control::{ + IlmRecoveryClassification, IlmRecoveryControlPage, IlmRecoveryControlView, IlmRecoveryProtocol, + inspect_recovery_control, list_recovery_controls, + }; + } + pub mod transition_transaction { pub use crate::bucket::lifecycle::transition_transaction::{ TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus, diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index a9957a25c..58e7ae607 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, tier_delete_journal, transition_transaction, + manual_transition_job, recovery_control, tier_delete_journal, transition_transaction, }; use crate::error::{Error, Result}; use crate::services::tier::tier_probe_intent; @@ -41,6 +41,7 @@ pub(crate) enum DurableIlmRecordKind { ManualTransitionScope, ManualTransitionTask, ManualTransitionWorkerResult, + RecoveryControl, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -105,8 +106,14 @@ pub(crate) const MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE: DurableIlmNamespace max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE, kind: DurableIlmRecordKind::ManualTransitionWorkerResult, }; +pub(crate) const RECOVERY_CONTROL_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "recovery-control", + prefix: recovery_control::ILM_RECOVERY_CONTROL_PREFIX, + max_record_size: recovery_control::MAX_ILM_RECOVERY_CONTROL_SIZE, + kind: DurableIlmRecordKind::RecoveryControl, +}; -pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [ +pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 10] = [ TIER_DELETE_JOURNAL_NAMESPACE, TIER_DELETE_JOURNAL_V6_NAMESPACE, TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE, @@ -116,6 +123,7 @@ pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [ MANUAL_TRANSITION_SCOPE_NAMESPACE, MANUAL_TRANSITION_TASK_NAMESPACE, MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE, + RECOVERY_CONTROL_NAMESPACE, ]; #[derive(Debug, Clone, PartialEq, Eq)] @@ -241,6 +249,18 @@ pub(crate) enum DurableIlmRecordCheckpoint { ManualTransitionWorkerResult { content_sha256: String, }, + RecoveryControl { + content_sha256: String, + identity_sha256: String, + source_generation_sha256: String, + first_seen_at_unix_nanos: i64, + revision: u64, + classification: recovery_control::IlmRecoveryClassification, + attempt_count: u64, + consecutive_failure_count: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + owner_fence_sha256: Option, + }, } impl DurableIlmRecordCheckpoint { @@ -254,7 +274,8 @@ impl DurableIlmRecordCheckpoint { | Self::ManualTransitionJob { content_sha256, .. } | Self::ManualTransitionScope { content_sha256, .. } | Self::ManualTransitionTask { content_sha256 } - | Self::ManualTransitionWorkerResult { content_sha256 } => content_sha256, + | Self::ManualTransitionWorkerResult { content_sha256 } + | Self::RecoveryControl { content_sha256, .. } => content_sha256, } } @@ -528,6 +549,51 @@ impl DurableIlmRecordCheckpoint { .. }, ) => previous_identity == next_identity && next_updated_at > previous_updated_at, + ( + Self::RecoveryControl { + identity_sha256: previous_identity, + source_generation_sha256: previous_generation, + first_seen_at_unix_nanos: previous_first_seen, + revision: previous_revision, + classification: previous_classification, + attempt_count: previous_attempts, + consecutive_failure_count: previous_failures, + owner_fence_sha256: previous_owner, + .. + }, + Self::RecoveryControl { + identity_sha256: next_identity, + source_generation_sha256: next_generation, + first_seen_at_unix_nanos: next_first_seen, + revision: next_revision, + classification: next_classification, + attempt_count: next_attempts, + consecutive_failure_count: next_failures, + owner_fence_sha256: next_owner, + .. + }, + ) => { + let adjacent = previous_identity == next_identity + && previous_first_seen == next_first_seen + && previous_revision.checked_add(1) == Some(*next_revision); + let claim = next_owner.is_some() + && *previous_classification == recovery_control::IlmRecoveryClassification::Retrying + && *next_classification == recovery_control::IlmRecoveryClassification::Retrying + && previous_attempts.checked_add(1) == Some(*next_attempts) + && previous_failures == next_failures; + let source_refresh = previous_owner.is_some() + && previous_owner == next_owner + && *previous_classification == recovery_control::IlmRecoveryClassification::Retrying + && *next_classification == recovery_control::IlmRecoveryClassification::Retrying + && previous_attempts == next_attempts + && previous_failures == next_failures + && previous_generation != next_generation; + let completion = previous_owner.is_some() + && next_owner.is_none() + && previous_generation == next_generation + && previous_attempts == next_attempts; + adjacent && (claim || source_refresh || completion) + } _ => false, }; @@ -553,6 +619,14 @@ impl DurableIlmRecordCheckpoint { { return false; } + if let Self::RecoveryControl { classification, .. } = terminal + && !matches!( + classification, + recovery_control::IlmRecoveryClassification::Terminal | recovery_control::IlmRecoveryClassification::Abandoned + ) + { + return false; + } if self == terminal || self.validate_successor(terminal).is_ok() { return true; } @@ -652,6 +726,32 @@ impl DurableIlmRecordCheckpoint { .is_some_and(|distance| tier_probe_state_reaches(*previous_state, *terminal_state, distance)) && (!previous_remote_version_known || previous_remote_version == terminal_remote_version) } + ( + Self::RecoveryControl { + identity_sha256: previous_identity, + source_generation_sha256: previous_generation, + first_seen_at_unix_nanos: previous_first_seen, + revision: previous_revision, + attempt_count: previous_attempts, + .. + }, + Self::RecoveryControl { + identity_sha256: terminal_identity, + source_generation_sha256: terminal_generation, + first_seen_at_unix_nanos: terminal_first_seen, + revision: terminal_revision, + attempt_count: terminal_attempts, + classification: + recovery_control::IlmRecoveryClassification::Terminal | recovery_control::IlmRecoveryClassification::Abandoned, + .. + }, + ) => { + previous_identity == terminal_identity + && (previous_generation == terminal_generation || terminal_attempts > previous_attempts) + && previous_first_seen == terminal_first_seen + && terminal_revision > previous_revision + && terminal_attempts >= previous_attempts + } _ => false, } } @@ -1219,6 +1319,35 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { + let (protocol, control_id) = recovery_control::recovery_control_id_from_record_object_name(path) + .map_err(|err| Error::other(err.to_string()))?; + let control = + recovery_control::IlmRecoveryControl::decode(&control_id, data).map_err(|err| Error::other(err.to_string()))?; + let canonical = recovery_control::recovery_control_record_object_name(protocol, &control_id) + .map_err(|err| Error::other(err.to_string()))?; + if canonical != path || control.identity.protocol != protocol { + return Err(Error::other("ILM recovery control path is not canonical")); + } + let identity_sha256 = checkpoint_hash(&control.identity)?; + let source_generation_sha256 = checkpoint_hash(&control.observed_source_generation)?; + let owner_fence_sha256 = control.owner.as_ref().map(checkpoint_hash).transpose()?; + ( + "control_id", + control_id, + DurableIlmRecordCheckpoint::RecoveryControl { + content_sha256, + identity_sha256, + source_generation_sha256, + first_seen_at_unix_nanos: control.first_seen_at_unix_nanos, + revision: control.revision, + classification: control.classification, + attempt_count: control.attempt_count, + consecutive_failure_count: control.consecutive_failure_count, + owner_fence_sha256, + }, + ) + } 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()))?; @@ -1412,6 +1541,94 @@ mod tests { .checkpoint } + fn recovery_control_fixture() -> recovery_control::IlmRecoveryControl { + let source_path = "ilm/transition-transactions/records/12/34/1234567890abcdef1234567890abcdef.json"; + let generation = recovery_control::IlmRecoverySourceGeneration::new( + transition_transaction::TRANSITION_TRANSACTION_SCHEMA, + "source-etag", + "a".repeat(64), + vec![recovery_control::IlmRecoverySourceCopy { + authority: "pool-0/set-0".to_string(), + canonical_path: source_path.to_string(), + etag: "source-etag".to_string(), + encoded_len: 128, + content_sha256: "a".repeat(64), + }], + ) + .expect("source generation should build"); + recovery_control::IlmRecoveryControl::new( + recovery_control::IlmRecoveryControlIdentity { + protocol: recovery_control::IlmRecoveryProtocol::TransitionTransaction, + canonical_source_path: source_path.to_string(), + stable_operation_identity: "12345678-90ab-cdef-1234-567890abcdef".to_string(), + record_class: "transition_transaction_v1".to_string(), + }, + generation, + recovery_control::IlmRecoveryClassification::Retrying, + 1_000_000_000, + recovery_control::IlmRecoveryErrorCode::None, + ) + .expect("recovery control should build") + } + + fn recovery_control_checkpoint(control: &recovery_control::IlmRecoveryControl) -> DurableIlmRecordCheckpoint { + let control_id = control.identity.source_operation_digest().expect("control id should derive"); + let path = recovery_control::recovery_control_record_object_name(control.identity.protocol, &control_id) + .expect("control path should build"); + let encoded = control.encode().expect("control should encode"); + let namespace = classify_durable_ilm_record(&path) + .expect("recovery control namespace should classify") + .expect("recovery control should be durable"); + assert_eq!(namespace, &RECOVERY_CONTROL_NAMESPACE); + validate_durable_ilm_record(&path, &encoded) + .expect("recovery control should validate") + .checkpoint + } + + #[test] + fn recovery_control_checkpoint_tracks_claim_retry_and_terminal_generations() { + let initial_control = recovery_control_fixture(); + let initial = recovery_control_checkpoint(&initial_control); + + let mut claimed_control = initial_control; + let mut advanced_generation = claimed_control.observed_source_generation.clone(); + advanced_generation.source_schema = "rustfs-transition-transaction-v2".to_string(); + claimed_control + .claim_for_source_generation("node-a", Uuid::new_v4(), 2_000_000_000, 300_000_000_000, advanced_generation) + .expect("control should claim"); + let claimed = recovery_control_checkpoint(&claimed_control); + initial.validate_successor(&claimed).expect("claim should advance receipt"); + + let mut retry_control = claimed_control; + retry_control + .record_retryable_failure(3_000_000_000, recovery_control::IlmRecoveryErrorCode::BackendTimeout) + .expect("retry should persist"); + let retry = recovery_control_checkpoint(&retry_control); + claimed.validate_successor(&retry).expect("retry should advance receipt"); + + let ready_at = retry_control + .next_attempt_at_unix_nanos + .expect("retry deadline should persist"); + let mut terminal_control = retry_control; + terminal_control + .claim("node-b", Uuid::new_v4(), ready_at, 300_000_000_000) + .expect("retry should claim"); + let reclaimed = recovery_control_checkpoint(&terminal_control); + retry.validate_successor(&reclaimed).expect("reclaim should advance receipt"); + terminal_control + .finish_attempt( + recovery_control::IlmRecoveryClassification::Terminal, + recovery_control::IlmRecoveryErrorCode::None, + ) + .expect("control should terminate"); + let terminal = recovery_control_checkpoint(&terminal_control); + reclaimed + .validate_successor(&terminal) + .expect("terminal state should advance receipt"); + assert!(initial.is_predecessor_of_terminal(&terminal)); + assert!(!initial.is_predecessor_of_terminal(&retry)); + } + #[test] fn tier_probe_intent_checkpoint_tracks_exact_monotonic_generations() { let initial_intent = tier_probe_intent_fixture(); diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index a2290c8ea..3eb93cca4 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -24,6 +24,7 @@ pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs, g mod object_handlers_common; mod object_lock_boundary; pub use self::core as lifecycle; +pub mod recovery_control; mod replication_sink; pub mod rule; mod runtime_boundary; diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_control.rs b/crates/ecstore/src/bucket/lifecycle/recovery_control.rs new file mode 100644 index 000000000..599849ca9 --- /dev/null +++ b/crates/ecstore/src/bucket/lifecycle/recovery_control.rs @@ -0,0 +1,1280 @@ +// 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 crate::disk::RUSTFS_META_BUCKET; +use crate::error::{Error, Result as EcstoreResult}; +use crate::object_api::ObjectOptions; +use crate::storage_api_contracts::{list::ListOperations as _, object::HTTPPreconditions}; +use crate::store::ECStore; + +pub const ILM_RECOVERY_CONTROL_SCHEMA: &str = "rustfs-ilm-recovery-control-v1"; +pub const ILM_RECOVERY_CONTROL_PREFIX: &str = "ilm/recovery-controls"; +pub const MAX_ILM_RECOVERY_CONTROL_SIZE: usize = 16 * 1024; +pub const MAX_RECOVERY_ATTEMPTS: u32 = 32; +const MAX_RECOVERY_AGE_NANOS: i64 = 7 * 24 * 60 * 60 * 1_000_000_000; +const MIN_RETRY_DELAY_NANOS: i64 = 60 * 1_000_000_000; +const MAX_RETRY_DELAY_NANOS: i64 = 60 * 60 * 1_000_000_000; + +pub type Result = std::result::Result; + +#[derive(Debug, thiserror::Error)] +pub enum IlmRecoveryControlError { + #[error("ILM recovery control is corrupt: {0}")] + Corrupt(&'static str), + #[error("ILM recovery control schema is unsupported: {0}")] + UnsupportedSchema(String), + #[error("ILM recovery control checksum mismatch")] + ChecksumMismatch, + #[error("ILM recovery control successor is invalid: {0}")] + InvalidSuccessor(&'static str), + #[error("ILM recovery control json error: {0}")] + Json(#[from] serde_json::Error), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryProtocol { + TransitionTransaction, + TierDeleteJournal, + TierDeleteManifest, +} + +impl IlmRecoveryProtocol { + pub fn as_str(self) -> &'static str { + match self { + Self::TransitionTransaction => "transition_transaction", + Self::TierDeleteJournal => "tier_delete_journal", + Self::TierDeleteManifest => "tier_delete_manifest", + } + } + + pub const fn all() -> [Self; 3] { + [Self::TransitionTransaction, Self::TierDeleteJournal, Self::TierDeleteManifest] + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryClassification { + Retrying, + RetainedAmbiguous, + Corrupt, + OperatorRequired, + Abandoned, + Terminal, +} + +impl IlmRecoveryClassification { + pub fn permits_automatic_attempt(self) -> bool { + self == Self::Retrying + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IlmRecoveryErrorCode { + None, + SourceUnavailable, + SourceDivergent, + SourceCorrupt, + SourceGenerationChanged, + BackendUnavailable, + BackendTimeout, + BackendThrottled, + BackendServerError, + AttemptLeaseExpired, + RemoteVersionUnknown, + RemoteProbeAmbiguous, + RemoteProbeUnsupported, + LocalCommitAmbiguous, + CasConflict, + CleanupFailed, + OperatorDispositionRequired, + UnsupportedSchema, + Unknown, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoverySourceCopy { + pub authority: String, + pub canonical_path: String, + pub etag: String, + pub encoded_len: u64, + pub content_sha256: String, +} + +impl IlmRecoverySourceCopy { + fn validate(&self) -> Result<()> { + if self.authority.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("source copy authority is empty")); + } + validate_canonical_source_path(&self.canonical_path)?; + if self.etag.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("source copy ETag is empty")); + } + if self.encoded_len == 0 { + return Err(IlmRecoveryControlError::Corrupt("source copy encoded length is zero")); + } + validate_sha256(&self.content_sha256, "source copy content checksum is invalid") + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoverySourceGeneration { + pub source_schema: String, + pub source_etag: String, + pub content_sha256: String, + pub copy_set_sha256: String, + pub copies: Vec, +} + +impl IlmRecoverySourceGeneration { + pub fn new( + source_schema: impl Into, + source_etag: impl Into, + content_sha256: impl Into, + mut copies: Vec, + ) -> Result { + copies.sort_by(|left, right| (&left.authority, &left.canonical_path).cmp(&(&right.authority, &right.canonical_path))); + let copy_set_sha256 = copy_set_digest(&copies)?; + let generation = Self { + source_schema: source_schema.into(), + source_etag: source_etag.into(), + content_sha256: content_sha256.into(), + copy_set_sha256, + copies, + }; + generation.validate()?; + Ok(generation) + } + + fn validate(&self) -> Result<()> { + if self.source_schema.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("source schema is empty")); + } + if self.source_etag.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("source ETag is empty")); + } + validate_sha256(&self.content_sha256, "source content checksum is invalid")?; + validate_sha256(&self.copy_set_sha256, "source copy-set checksum is invalid")?; + if self.copies.is_empty() { + return Err(IlmRecoveryControlError::Corrupt("source copy set is empty")); + } + for copy in &self.copies { + copy.validate()?; + } + if !self + .copies + .windows(2) + .all(|pair| (&pair[0].authority, &pair[0].canonical_path) < (&pair[1].authority, &pair[1].canonical_path)) + { + return Err(IlmRecoveryControlError::Corrupt("source copies are not in unique canonical order")); + } + if copy_set_digest(&self.copies)? != self.copy_set_sha256 { + return Err(IlmRecoveryControlError::Corrupt("source copy-set checksum does not match copies")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryControlIdentity { + pub protocol: IlmRecoveryProtocol, + pub canonical_source_path: String, + pub stable_operation_identity: String, + pub record_class: String, +} + +impl IlmRecoveryControlIdentity { + pub fn source_operation_digest(&self) -> Result { + self.validate()?; + Ok(length_delimited_digest(&[ + self.protocol.as_str().as_bytes(), + self.canonical_source_path.as_bytes(), + self.stable_operation_identity.as_bytes(), + ])) + } + + fn validate(&self) -> Result<()> { + validate_canonical_source_path(&self.canonical_source_path)?; + if self.stable_operation_identity.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("stable operation identity is empty")); + } + if self.record_class.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("record class is empty")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryOwnerLease { + pub owner_id: String, + pub owner_epoch: Uuid, + pub lease_acquired_at_unix_nanos: i64, + pub lease_expires_at_unix_nanos: i64, +} + +impl IlmRecoveryOwnerLease { + fn validate(&self) -> Result<()> { + if self.owner_id.trim().is_empty() { + return Err(IlmRecoveryControlError::Corrupt("owner id is empty")); + } + if self.owner_epoch.is_nil() { + return Err(IlmRecoveryControlError::Corrupt("owner epoch is nil")); + } + if self.lease_acquired_at_unix_nanos <= 0 { + return Err(IlmRecoveryControlError::Corrupt("lease acquisition timestamp is not positive")); + } + if self.lease_expires_at_unix_nanos <= self.lease_acquired_at_unix_nanos { + return Err(IlmRecoveryControlError::Corrupt("lease expiry does not follow acquisition")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IlmRecoveryControl { + pub identity: IlmRecoveryControlIdentity, + pub first_seen_at_unix_nanos: i64, + pub observed_source_generation: IlmRecoverySourceGeneration, + pub revision: u64, + pub classification: IlmRecoveryClassification, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub owner: Option, + pub attempt_count: u64, + pub consecutive_failure_count: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub first_failure_at_unix_nanos: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_failure_at_unix_nanos: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_attempt_at_unix_nanos: Option, + pub last_error_code: IlmRecoveryErrorCode, +} + +impl IlmRecoveryControl { + pub fn new( + identity: IlmRecoveryControlIdentity, + observed_source_generation: IlmRecoverySourceGeneration, + classification: IlmRecoveryClassification, + now_unix_nanos: i64, + last_error_code: IlmRecoveryErrorCode, + ) -> Result { + let control = Self { + identity, + first_seen_at_unix_nanos: now_unix_nanos, + observed_source_generation, + revision: 1, + classification, + owner: None, + attempt_count: 0, + consecutive_failure_count: 0, + first_failure_at_unix_nanos: None, + last_failure_at_unix_nanos: None, + next_attempt_at_unix_nanos: None, + last_error_code, + }; + control.validate()?; + Ok(control) + } + + pub fn validate(&self) -> Result<()> { + self.identity.validate()?; + self.observed_source_generation.validate()?; + if self.first_seen_at_unix_nanos <= 0 { + return Err(IlmRecoveryControlError::Corrupt("first-seen timestamp is not positive")); + } + if self.revision == 0 { + return Err(IlmRecoveryControlError::Corrupt("revision is zero")); + } + if let Some(owner) = &self.owner { + owner.validate()?; + if !self.classification.permits_automatic_attempt() { + return Err(IlmRecoveryControlError::Corrupt("non-retrying control carries an owner lease")); + } + } + match ( + self.consecutive_failure_count, + self.first_failure_at_unix_nanos, + self.last_failure_at_unix_nanos, + self.next_attempt_at_unix_nanos, + ) { + (0, None, None, None) => {} + (0, Some(first), Some(last), None) if first > 0 && last >= first => {} + (0, _, _, _) => return Err(IlmRecoveryControlError::Corrupt("zero failures carry inconsistent history")), + (_, Some(first), Some(last), next) if first > 0 && last >= first => { + if self.classification == IlmRecoveryClassification::Retrying && next.is_none_or(|next| next <= last) { + return Err(IlmRecoveryControlError::Corrupt("retrying control has no future retry timestamp")); + } + } + _ => return Err(IlmRecoveryControlError::Corrupt("failure counters and timestamps are inconsistent")), + } + if u64::from(self.consecutive_failure_count) > self.attempt_count { + return Err(IlmRecoveryControlError::Corrupt("consecutive failures exceed lifetime attempts")); + } + if self.classification == IlmRecoveryClassification::Retrying + && self.last_error_code == IlmRecoveryErrorCode::None + && self.consecutive_failure_count > 0 + { + return Err(IlmRecoveryControlError::Corrupt("failed retry has no bounded error code")); + } + Ok(()) + } + + pub fn should_attempt_at(&self, now_unix_nanos: i64) -> bool { + self.classification.permits_automatic_attempt() + && self + .owner + .as_ref() + .is_none_or(|owner| owner.lease_expires_at_unix_nanos <= now_unix_nanos) + && self.next_attempt_at_unix_nanos.is_none_or(|next| next <= now_unix_nanos) + } + + pub fn claim( + &mut self, + owner_id: impl Into, + owner_epoch: Uuid, + now_unix_nanos: i64, + lease_duration_nanos: i64, + ) -> Result<()> { + self.claim_for_source_generation( + owner_id, + owner_epoch, + now_unix_nanos, + lease_duration_nanos, + self.observed_source_generation.clone(), + ) + } + + pub fn claim_for_source_generation( + &mut self, + owner_id: impl Into, + owner_epoch: Uuid, + now_unix_nanos: i64, + lease_duration_nanos: i64, + observed_source_generation: IlmRecoverySourceGeneration, + ) -> Result<()> { + if !self.should_attempt_at(now_unix_nanos) { + return Err(IlmRecoveryControlError::InvalidSuccessor("control is not ready for an attempt")); + } + let lease_expires_at_unix_nanos = now_unix_nanos + .checked_add(lease_duration_nanos) + .ok_or(IlmRecoveryControlError::Corrupt("lease timestamp overflow"))?; + self.bump_revision()?; + self.observed_source_generation = observed_source_generation; + self.owner = Some(IlmRecoveryOwnerLease { + owner_id: owner_id.into(), + owner_epoch, + lease_acquired_at_unix_nanos: now_unix_nanos, + lease_expires_at_unix_nanos, + }); + self.attempt_count = self + .attempt_count + .checked_add(1) + .ok_or(IlmRecoveryControlError::Corrupt("attempt count overflow"))?; + self.validate() + } + + pub fn record_retryable_failure(&mut self, now_unix_nanos: i64, code: IlmRecoveryErrorCode) -> Result<()> { + if code == IlmRecoveryErrorCode::None { + return Err(IlmRecoveryControlError::InvalidSuccessor("retryable failure requires an error code")); + } + let attempt_started_at = self + .owner + .as_ref() + .ok_or(IlmRecoveryControlError::InvalidSuccessor("retryable failure requires an owner lease"))? + .lease_acquired_at_unix_nanos; + if now_unix_nanos < attempt_started_at { + return Err(IlmRecoveryControlError::InvalidSuccessor("retryable failure predates its owner claim")); + } + let failures = self + .consecutive_failure_count + .checked_add(1) + .ok_or(IlmRecoveryControlError::Corrupt("failure count overflow"))?; + let first_failure_at = self.first_failure_at_unix_nanos.unwrap_or(attempt_started_at); + let age = now_unix_nanos.saturating_sub(first_failure_at); + self.bump_revision()?; + self.owner = None; + self.consecutive_failure_count = failures; + self.first_failure_at_unix_nanos = Some(first_failure_at); + self.last_failure_at_unix_nanos = Some(now_unix_nanos); + self.last_error_code = code; + if failures >= MAX_RECOVERY_ATTEMPTS + || self.attempt_count >= u64::from(MAX_RECOVERY_ATTEMPTS) + || age >= MAX_RECOVERY_AGE_NANOS + { + self.classification = IlmRecoveryClassification::OperatorRequired; + self.next_attempt_at_unix_nanos = None; + } else { + self.classification = IlmRecoveryClassification::Retrying; + self.next_attempt_at_unix_nanos = Some( + now_unix_nanos + .checked_add(retry_delay_nanos( + &self.observed_source_generation.copy_set_sha256, + self.attempt_count, + failures, + )) + .ok_or(IlmRecoveryControlError::Corrupt("retry timestamp overflow"))?, + ); + } + self.validate() + } + + pub fn record_expired_attempt(&mut self, now_unix_nanos: i64) -> Result<()> { + let lease_expires_at = self + .owner + .as_ref() + .ok_or(IlmRecoveryControlError::InvalidSuccessor("expired attempt requires an owner lease"))? + .lease_expires_at_unix_nanos; + if lease_expires_at > now_unix_nanos { + return Err(IlmRecoveryControlError::InvalidSuccessor("attempt owner lease is still active")); + } + self.record_retryable_failure(now_unix_nanos, IlmRecoveryErrorCode::AttemptLeaseExpired) + } + + pub fn refresh_owned_source_generation(&mut self, observed_source_generation: IlmRecoverySourceGeneration) -> Result<()> { + if self.owner.is_none() || self.classification != IlmRecoveryClassification::Retrying { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "source generation refresh requires a retrying owner", + )); + } + self.bump_revision()?; + self.observed_source_generation = observed_source_generation; + self.validate() + } + + pub fn finish_attempt(&mut self, classification: IlmRecoveryClassification, code: IlmRecoveryErrorCode) -> Result<()> { + if self.owner.is_none() { + return Err(IlmRecoveryControlError::InvalidSuccessor("finishing an attempt requires an owner lease")); + } + if classification == IlmRecoveryClassification::Retrying { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "successful attempt result cannot remain retrying", + )); + } + self.bump_revision()?; + self.owner = None; + self.classification = classification; + self.consecutive_failure_count = 0; + self.next_attempt_at_unix_nanos = None; + self.last_error_code = code; + self.validate() + } + + pub fn validate_successor(&self, next: &Self) -> Result<()> { + self.validate()?; + next.validate()?; + if self.identity != next.identity || self.first_seen_at_unix_nanos != next.first_seen_at_unix_nanos { + return Err(IlmRecoveryControlError::InvalidSuccessor("immutable identity changed")); + } + if self.revision.checked_add(1) != Some(next.revision) { + return Err(IlmRecoveryControlError::InvalidSuccessor("revision did not advance by one")); + } + match (&self.owner, &next.owner) { + (Some(current_owner), Some(next_owner)) if current_owner == next_owner => { + self.validate_source_refresh_successor(next) + } + (_, Some(_)) => self.validate_claim_successor(next), + (Some(_), None) + if self + .consecutive_failure_count + .checked_add(1) + .is_some_and(|failures| next.consecutive_failure_count == failures) => + { + self.validate_failure_successor(next) + } + (Some(_), None) => self.validate_finish_successor(next), + (None, None) => Err(IlmRecoveryControlError::InvalidSuccessor( + "ownerless control cannot advance without a claim", + )), + } + } + + fn validate_claim_successor(&self, next: &Self) -> Result<()> { + if self.classification != IlmRecoveryClassification::Retrying + || next.classification != IlmRecoveryClassification::Retrying + || self + .attempt_count + .checked_add(1) + .is_none_or(|attempts| next.attempt_count != attempts) + || next.consecutive_failure_count != self.consecutive_failure_count + || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos + || next.last_failure_at_unix_nanos != self.last_failure_at_unix_nanos + || next.next_attempt_at_unix_nanos != self.next_attempt_at_unix_nanos + || next.last_error_code != self.last_error_code + { + return Err(IlmRecoveryControlError::InvalidSuccessor("claim changed non-owner recovery state")); + } + Ok(()) + } + + fn validate_source_refresh_successor(&self, next: &Self) -> Result<()> { + if self.classification != IlmRecoveryClassification::Retrying + || next.classification != IlmRecoveryClassification::Retrying + || next.attempt_count != self.attempt_count + || next.consecutive_failure_count != self.consecutive_failure_count + || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos + || next.last_failure_at_unix_nanos != self.last_failure_at_unix_nanos + || next.next_attempt_at_unix_nanos != self.next_attempt_at_unix_nanos + || next.last_error_code != self.last_error_code + { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "source generation refresh changed non-source recovery state", + )); + } + Ok(()) + } + + fn validate_failure_successor(&self, next: &Self) -> Result<()> { + let owner = self + .owner + .as_ref() + .ok_or(IlmRecoveryControlError::InvalidSuccessor("retry failure has no owner claim"))?; + if next.observed_source_generation != self.observed_source_generation + || next.attempt_count != self.attempt_count + || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos.or(Some(owner.lease_acquired_at_unix_nanos)) + || next.last_error_code == IlmRecoveryErrorCode::None + { + return Err(IlmRecoveryControlError::InvalidSuccessor("retry failure changed immutable attempt state")); + } + let last_failure = next + .last_failure_at_unix_nanos + .ok_or(IlmRecoveryControlError::InvalidSuccessor("retry failure has no timestamp"))?; + let first_failure = next + .first_failure_at_unix_nanos + .ok_or(IlmRecoveryControlError::InvalidSuccessor("retry failure has no first timestamp"))?; + if self + .owner + .as_ref() + .is_none_or(|owner| last_failure < owner.lease_acquired_at_unix_nanos) + { + return Err(IlmRecoveryControlError::InvalidSuccessor("retry failure predates its owner claim")); + } + let exhausted = next.consecutive_failure_count >= MAX_RECOVERY_ATTEMPTS + || next.attempt_count >= u64::from(MAX_RECOVERY_ATTEMPTS) + || last_failure.saturating_sub(first_failure) >= MAX_RECOVERY_AGE_NANOS; + let expected_next = if exhausted { + None + } else { + Some( + last_failure + .checked_add(retry_delay_nanos( + &next.observed_source_generation.copy_set_sha256, + next.attempt_count, + next.consecutive_failure_count, + )) + .ok_or(IlmRecoveryControlError::InvalidSuccessor("retry timestamp overflowed"))?, + ) + }; + if next.classification + != if exhausted { + IlmRecoveryClassification::OperatorRequired + } else { + IlmRecoveryClassification::Retrying + } + || next.next_attempt_at_unix_nanos != expected_next + { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "retry failure has an invalid terminal or backoff state", + )); + } + Ok(()) + } + + fn validate_finish_successor(&self, next: &Self) -> Result<()> { + if next.observed_source_generation != self.observed_source_generation + || next.attempt_count != self.attempt_count + || next.classification == IlmRecoveryClassification::Retrying + || next.consecutive_failure_count != 0 + || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos + || next.last_failure_at_unix_nanos != self.last_failure_at_unix_nanos + || next.next_attempt_at_unix_nanos.is_some() + { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "finished attempt changed immutable recovery state", + )); + } + Ok(()) + } + + pub fn encode(&self) -> Result> { + self.validate()?; + let control_bytes = serde_json::to_vec(self)?; + let persisted = PersistedIlmRecoveryControl { + schema: ILM_RECOVERY_CONTROL_SCHEMA.to_string(), + content_sha256: hex_sha256(&control_bytes, ToOwned::to_owned), + control: self.clone(), + }; + let encoded = serde_json::to_vec(&persisted)?; + if encoded.len() > MAX_ILM_RECOVERY_CONTROL_SIZE { + return Err(IlmRecoveryControlError::Corrupt("encoded control exceeds maximum size")); + } + Ok(encoded) + } + + pub fn decode(expected_control_id: &str, data: &[u8]) -> Result { + validate_sha256(expected_control_id, "control id is invalid")?; + if data.len() > MAX_ILM_RECOVERY_CONTROL_SIZE { + return Err(IlmRecoveryControlError::Corrupt("encoded control exceeds maximum size")); + } + let persisted: PersistedIlmRecoveryControl = serde_json::from_slice(data)?; + if persisted.schema != ILM_RECOVERY_CONTROL_SCHEMA { + return Err(IlmRecoveryControlError::UnsupportedSchema(persisted.schema)); + } + validate_sha256(&persisted.content_sha256, "content checksum is invalid")?; + let control_bytes = serde_json::to_vec(&persisted.control)?; + if hex_sha256(&control_bytes, ToOwned::to_owned) != persisted.content_sha256 { + return Err(IlmRecoveryControlError::ChecksumMismatch); + } + if persisted.control.identity.source_operation_digest()? != expected_control_id { + return Err(IlmRecoveryControlError::Corrupt("control id does not match record key")); + } + persisted.control.validate()?; + Ok(persisted.control) + } + + fn bump_revision(&mut self) -> Result<()> { + self.revision = self + .revision + .checked_add(1) + .ok_or(IlmRecoveryControlError::Corrupt("revision overflow"))?; + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObservedIlmRecoveryControl { + pub control: IlmRecoveryControl, + pub etag: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObservedIlmRecoverySource { + pub generation: IlmRecoverySourceGeneration, + pub canonical_data: Option>, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct IlmRecoveryControlView { + pub control_id: String, + pub protocol: IlmRecoveryProtocol, + pub classification: IlmRecoveryClassification, + pub schema: &'static str, + pub revision: u64, + pub attempt_count: u64, + pub consecutive_failure_count: u32, + pub first_seen_at_unix_nanos: i64, + pub first_failure_at_unix_nanos: Option, + pub last_failure_at_unix_nanos: Option, + pub next_attempt_at_unix_nanos: Option, + pub last_error_code: IlmRecoveryErrorCode, + pub source_schema: String, + pub source_generation_sha256: String, + pub copy_set_sha256: String, + pub source_copy_count: usize, +} + +impl IlmRecoveryControlView { + fn from_control(control_id: String, control: &IlmRecoveryControl) -> EcstoreResult { + let generation = serde_json::to_vec(&control.observed_source_generation).map_err(Error::other)?; + Ok(Self { + control_id, + protocol: control.identity.protocol, + classification: control.classification, + schema: ILM_RECOVERY_CONTROL_SCHEMA, + revision: control.revision, + attempt_count: control.attempt_count, + consecutive_failure_count: control.consecutive_failure_count, + first_seen_at_unix_nanos: control.first_seen_at_unix_nanos, + first_failure_at_unix_nanos: control.first_failure_at_unix_nanos, + last_failure_at_unix_nanos: control.last_failure_at_unix_nanos, + next_attempt_at_unix_nanos: control.next_attempt_at_unix_nanos, + last_error_code: control.last_error_code, + source_schema: control.observed_source_generation.source_schema.clone(), + source_generation_sha256: hex_sha256(&generation, ToOwned::to_owned), + copy_set_sha256: control.observed_source_generation.copy_set_sha256.clone(), + source_copy_count: control.observed_source_generation.copies.len(), + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct IlmRecoveryControlPage { + pub records: Vec, + pub next_marker: Option, + pub truncated: bool, + pub incomplete: bool, +} + +impl ObservedIlmRecoverySource { + pub fn is_consistent(&self) -> bool { + self.canonical_data.is_some() + && self.generation.copies.iter().all(|copy| { + copy.etag == self.generation.source_etag + && copy.content_sha256 == self.generation.content_sha256 + && copy.encoded_len + == self + .canonical_data + .as_ref() + .map_or(0, |data| u64::try_from(data.len()).unwrap_or(u64::MAX)) + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedIlmRecoveryControl { + schema: String, + content_sha256: String, + control: IlmRecoveryControl, +} + +pub fn recovery_control_record_object_name(protocol: IlmRecoveryProtocol, control_id: &str) -> Result { + validate_sha256(control_id, "control id is invalid")?; + Ok(format!( + "{}/{}/{}/{}/{}.json", + ILM_RECOVERY_CONTROL_PREFIX, + protocol.as_str(), + &control_id[..2], + &control_id[2..4], + control_id + )) +} + +pub fn recovery_control_id_from_record_object_name(object: &str) -> Result<(IlmRecoveryProtocol, String)> { + let suffix = object + .strip_prefix(ILM_RECOVERY_CONTROL_PREFIX) + .and_then(|suffix| suffix.strip_prefix('/')) + .ok_or(IlmRecoveryControlError::Corrupt("control record path has wrong prefix"))?; + let mut parts = suffix.split('/'); + let protocol = match parts.next() { + Some("transition_transaction") => IlmRecoveryProtocol::TransitionTransaction, + Some("tier_delete_journal") => IlmRecoveryProtocol::TierDeleteJournal, + Some("tier_delete_manifest") => IlmRecoveryProtocol::TierDeleteManifest, + _ => return Err(IlmRecoveryControlError::Corrupt("control record protocol is invalid")), + }; + let shard_a = parts + .next() + .ok_or(IlmRecoveryControlError::Corrupt("control record path is incomplete"))?; + let shard_b = parts + .next() + .ok_or(IlmRecoveryControlError::Corrupt("control record path is incomplete"))?; + let control_id = parts + .next() + .and_then(|name| name.strip_suffix(".json")) + .ok_or(IlmRecoveryControlError::Corrupt("control record suffix is invalid"))?; + if parts.next().is_some() { + return Err(IlmRecoveryControlError::Corrupt("control record path is not canonical")); + } + validate_sha256(control_id, "control id is invalid")?; + if shard_a != &control_id[..2] || shard_b != &control_id[2..4] { + return Err(IlmRecoveryControlError::Corrupt("control record shard does not match control id")); + } + Ok((protocol, control_id.to_string())) +} + +pub async fn save_recovery_control_if_absent(api: Arc, control: &IlmRecoveryControl) -> EcstoreResult<()> { + let control_id = control + .identity + .source_operation_digest() + .map_err(recovery_control_store_error)?; + let object = + recovery_control_record_object_name(control.identity.protocol, &control_id).map_err(recovery_control_store_error)?; + let data = control.encode().map_err(recovery_control_store_error)?; + config_boundary::save_config_with_opts( + api.clone(), + &object, + data.clone(), + &ObjectOptions { + max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await?; + api.record_durable_ilm_decommission_progress(&object, &data).await +} + +pub async fn observe_recovery_source( + api: Arc, + canonical_path: &str, + source_schema: &str, +) -> EcstoreResult { + validate_canonical_source_path(canonical_path).map_err(recovery_control_store_error)?; + if source_schema.trim().is_empty() { + return Err(Error::other("ILM recovery source schema is empty")); + } + + let mut copies = Vec::new(); + let mut observations = Vec::new(); + for set in api.all_set_disks() { + let authority = format!("pool-{}/set-{}", set.pool_index, set.set_index); + match config_boundary::read_config_with_metadata(set, canonical_path, &ObjectOptions::default()).await { + Ok((data, metadata)) => { + let etag = metadata + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("ILM recovery source copy is missing an ETag"))?; + let encoded_len = + u64::try_from(data.len()).map_err(|_| Error::other("ILM recovery source copy length does not fit u64"))?; + let content_sha256 = hex_sha256(&data, ToOwned::to_owned); + copies.push(IlmRecoverySourceCopy { + authority, + canonical_path: canonical_path.to_string(), + etag: etag.clone(), + encoded_len, + content_sha256: content_sha256.clone(), + }); + observations.push((etag, content_sha256, data)); + } + Err(err) if recovery_source_is_missing(&err) => {} + Err(err) => return Err(err), + } + } + let Some((source_etag, content_sha256, first_data)) = observations.first().cloned() else { + return Err(Error::ConfigNotFound); + }; + let consistent = observations + .iter() + .all(|(etag, digest, data)| etag == &source_etag && digest == &content_sha256 && data == &first_data); + let generation = IlmRecoverySourceGeneration::new(source_schema, source_etag, content_sha256, copies) + .map_err(recovery_control_store_error)?; + Ok(ObservedIlmRecoverySource { + generation, + canonical_data: consistent.then_some(first_data), + }) +} + +pub async fn load_recovery_control( + api: Arc, + protocol: IlmRecoveryProtocol, + control_id: &str, +) -> EcstoreResult { + let object = recovery_control_record_object_name(protocol, control_id).map_err(recovery_control_store_error)?; + let (data, metadata) = config_boundary::read_config_with_metadata(api, &object, &ObjectOptions::default()).await?; + let etag = metadata + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("ILM recovery control is missing an ETag"))?; + let control = IlmRecoveryControl::decode(control_id, &data).map_err(recovery_control_store_error)?; + if control.identity.protocol != protocol { + return Err(Error::other("ILM recovery control protocol does not match record path")); + } + Ok(ObservedIlmRecoveryControl { control, etag }) +} + +pub async fn inspect_recovery_control(api: Arc, control_id: &str) -> EcstoreResult { + validate_sha256(control_id, "control id is invalid").map_err(recovery_control_store_error)?; + for protocol in IlmRecoveryProtocol::all() { + match load_recovery_control(api.clone(), protocol, control_id).await { + Ok(observed) => return IlmRecoveryControlView::from_control(control_id.to_string(), &observed.control), + Err(Error::ConfigNotFound) => {} + Err(err) => return Err(err), + } + } + Err(Error::ConfigNotFound) +} + +pub async fn list_recovery_controls( + api: Arc, + protocol: IlmRecoveryProtocol, + classification: Option, + limit: usize, + marker: Option, +) -> EcstoreResult { + if !(1..=1_000).contains(&limit) { + return Err(Error::other("ILM recovery control list limit must be between 1 and 1000")); + } + let prefix = format!("{}/{}/", ILM_RECOVERY_CONTROL_PREFIX, protocol.as_str()); + let page = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + &prefix, + marker, + None, + i32::try_from(limit).unwrap_or(1_000), + false, + None, + false, + ) + .await?; + if page.is_truncated && page.next_continuation_token.is_none() { + return Err(Error::other( + "ILM recovery control list returned a truncated page without a continuation marker", + )); + } + + let mut records = Vec::new(); + let mut incomplete = false; + for object in page.objects { + let parsed = recovery_control_id_from_record_object_name(&object.name); + let (path_protocol, control_id) = match parsed { + Ok(parsed) if parsed.0 == protocol => parsed, + Ok(_) | Err(_) => { + incomplete = true; + continue; + } + }; + match load_recovery_control(api.clone(), path_protocol, &control_id).await { + Ok(observed) if classification.is_none_or(|filter| observed.control.classification == filter) => { + records.push(IlmRecoveryControlView::from_control(control_id, &observed.control)?); + } + Ok(_) => {} + Err(Error::ConfigNotFound) => {} + Err(_) => incomplete = true, + } + } + + Ok(IlmRecoveryControlPage { + records, + next_marker: page.next_continuation_token, + truncated: page.is_truncated, + incomplete, + }) +} + +pub async fn save_recovery_control_if_current( + api: Arc, + current: &ObservedIlmRecoveryControl, + next: &IlmRecoveryControl, +) -> EcstoreResult<()> { + current + .control + .validate_successor(next) + .map_err(recovery_control_store_error)?; + let control_id = current + .control + .identity + .source_operation_digest() + .map_err(recovery_control_store_error)?; + let authoritative = load_recovery_control(api.clone(), current.control.identity.protocol, &control_id).await?; + if &authoritative != current { + return Err(Error::PreconditionFailed); + } + let object = recovery_control_record_object_name(current.control.identity.protocol, &control_id) + .map_err(recovery_control_store_error)?; + let data = next.encode().map_err(recovery_control_store_error)?; + config_boundary::save_config_with_opts( + api.clone(), + &object, + data.clone(), + &ObjectOptions { + max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(current.etag.clone()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await?; + api.record_durable_ilm_decommission_progress(&object, &data).await +} + +fn retry_delay_nanos(copy_set_sha256: &str, attempt_count: u64, consecutive_failure_count: u32) -> i64 { + let exponent = consecutive_failure_count.saturating_sub(1).min(6); + let base = MIN_RETRY_DELAY_NANOS + .saturating_mul(1_i64 << exponent) + .min(MAX_RETRY_DELAY_NANOS); + let seed = length_delimited_digest(&[copy_set_sha256.as_bytes(), &attempt_count.to_be_bytes()]); + let jitter_bucket = u8::from_str_radix(&seed[..2], 16).unwrap_or(0) % 21; + base.saturating_mul(i64::from(80 + jitter_bucket)) / 100 +} + +fn copy_set_digest(copies: &[IlmRecoverySourceCopy]) -> Result { + let encoded = serde_json::to_vec(copies)?; + Ok(hex_sha256(&encoded, ToOwned::to_owned)) +} + +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 validate_canonical_source_path(path: &str) -> Result<()> { + if path.is_empty() || path.starts_with('/') || path.ends_with('/') || path.split('/').any(|part| part.is_empty()) { + return Err(IlmRecoveryControlError::Corrupt("canonical source path is invalid")); + } + Ok(()) +} + +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(IlmRecoveryControlError::Corrupt(message)); + } + Ok(()) +} + +fn recovery_control_store_error(err: IlmRecoveryControlError) -> Error { + Error::other(err) +} + +fn recovery_source_is_missing(err: &Error) -> bool { + matches!( + err, + Error::ConfigNotFound | Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::VersionNotFound(_, _, _) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + const SOURCE_PATH: &str = "ilm/transition-transactions/records/12/34/1234567890abcdef1234567890abcdef.json"; + + fn generation() -> IlmRecoverySourceGeneration { + let content_sha256 = hex_sha256(b"source", ToOwned::to_owned); + IlmRecoverySourceGeneration::new( + "rustfs-transition-transaction-v1", + "etag-a", + content_sha256.clone(), + vec![ + IlmRecoverySourceCopy { + authority: "pool-1/set-0".to_string(), + canonical_path: SOURCE_PATH.to_string(), + etag: "etag-b".to_string(), + encoded_len: 6, + content_sha256: content_sha256.clone(), + }, + IlmRecoverySourceCopy { + authority: "pool-0/set-1".to_string(), + canonical_path: SOURCE_PATH.to_string(), + etag: "etag-a".to_string(), + encoded_len: 6, + content_sha256, + }, + ], + ) + .expect("source generation should build") + } + + fn control() -> IlmRecoveryControl { + IlmRecoveryControl::new( + IlmRecoveryControlIdentity { + protocol: IlmRecoveryProtocol::TransitionTransaction, + canonical_source_path: SOURCE_PATH.to_string(), + stable_operation_identity: "12345678-90ab-cdef-1234-567890abcdef".to_string(), + record_class: "transition_transaction_v1".to_string(), + }, + generation(), + IlmRecoveryClassification::Retrying, + 1_000_000_000, + IlmRecoveryErrorCode::None, + ) + .expect("control should build") + } + + #[test] + fn recovery_control_round_trip_and_canonical_path() { + let control = control(); + let control_id = control.identity.source_operation_digest().expect("control id should derive"); + let path = + recovery_control_record_object_name(control.identity.protocol, &control_id).expect("control path should build"); + assert_eq!( + recovery_control_id_from_record_object_name(&path).expect("control path should parse"), + (control.identity.protocol, control_id.clone()) + ); + let encoded = control.encode().expect("control should encode"); + assert_eq!(IlmRecoveryControl::decode(&control_id, &encoded).expect("control should decode"), control); + } + + #[test] + fn recovery_control_rejects_noncanonical_copy_set_and_tampering() { + let mut noncanonical = control(); + noncanonical.observed_source_generation.copies.swap(0, 1); + assert!(matches!(noncanonical.validate(), Err(IlmRecoveryControlError::Corrupt(_)))); + + let control = control(); + let control_id = control.identity.source_operation_digest().expect("control id should derive"); + let mut persisted: serde_json::Value = + serde_json::from_slice(&control.encode().expect("control should encode")).expect("encoded control should be json"); + persisted["control"]["attempt_count"] = serde_json::json!(9); + let tampered = serde_json::to_vec(&persisted).expect("tampered json should encode"); + assert!(matches!( + IlmRecoveryControl::decode(&control_id, &tampered), + Err(IlmRecoveryControlError::ChecksumMismatch) + )); + } + + #[test] + fn recovery_control_persists_deterministic_bounded_backoff() { + let mut first = control(); + let mut second = first.clone(); + for control in [&mut first, &mut second] { + control + .claim("node-a", Uuid::new_v4(), 2_000_000_000, 300_000_000_000) + .expect("attempt should claim"); + control + .record_retryable_failure(3_000_000_000, IlmRecoveryErrorCode::BackendTimeout) + .expect("failure should schedule retry"); + } + assert_eq!(first.next_attempt_at_unix_nanos, second.next_attempt_at_unix_nanos); + let delay = first.next_attempt_at_unix_nanos.expect("retry time") - 3_000_000_000; + assert!((48_000_000_000..=60_000_000_000).contains(&delay)); + assert!(!first.should_attempt_at(first.next_attempt_at_unix_nanos.expect("retry time") - 1)); + assert!(first.should_attempt_at(first.next_attempt_at_unix_nanos.expect("retry time"))); + } + + #[test] + fn recovery_control_stops_after_bounded_failures() { + let mut control = control(); + let mut now = 2_000_000_000; + for _ in 0..MAX_RECOVERY_ATTEMPTS { + let ready = control.next_attempt_at_unix_nanos.unwrap_or(now); + now = now.max(ready); + control + .claim("node-a", Uuid::new_v4(), now, 300_000_000_000) + .expect("attempt should claim"); + control + .record_retryable_failure(now + 1, IlmRecoveryErrorCode::BackendTimeout) + .expect("failure should persist"); + now += 2; + } + assert_eq!(control.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(control.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS)); + assert_eq!(control.next_attempt_at_unix_nanos, None); + } + + #[test] + fn recovery_control_expired_timeout_and_cancellation_attempts_stop_at_bounds() { + let mut active = control(); + active + .claim("node-a", Uuid::new_v4(), 2_000_000_000, 2) + .expect("active attempt should claim"); + assert!(matches!( + active.record_expired_attempt(2_000_000_001), + Err(IlmRecoveryControlError::InvalidSuccessor("attempt owner lease is still active")) + )); + + let mut bounded = control(); + let mut now = 2_000_000_000; + for attempt in 1..=MAX_RECOVERY_ATTEMPTS { + now = now.max(bounded.next_attempt_at_unix_nanos.unwrap_or(now)); + bounded + .claim("node-a", Uuid::new_v4(), now, 1) + .expect("timeout or cancellation attempt should claim"); + let abandonment = if attempt % 2 == 0 { "cancellation" } else { "timeout" }; + bounded + .record_expired_attempt(now + 1) + .unwrap_or_else(|err| panic!("expired {abandonment} attempt should consume its budget: {err}")); + if attempt < MAX_RECOVERY_ATTEMPTS { + assert_eq!(bounded.classification, IlmRecoveryClassification::Retrying); + } + now += 2; + } + + assert_eq!(bounded.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(bounded.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS)); + assert_eq!(bounded.consecutive_failure_count, MAX_RECOVERY_ATTEMPTS); + assert_eq!(bounded.last_error_code, IlmRecoveryErrorCode::AttemptLeaseExpired); + assert_eq!(bounded.next_attempt_at_unix_nanos, None); + + let mut younger = control(); + younger + .claim("node-a", Uuid::new_v4(), now, 1) + .expect("younger attempt should claim"); + younger + .record_expired_attempt(now + MAX_RECOVERY_AGE_NANOS - 1) + .expect("younger expired attempt should be recorded"); + assert_eq!(younger.classification, IlmRecoveryClassification::Retrying); + + let mut aged = control(); + aged.claim("node-a", Uuid::new_v4(), now, 1) + .expect("aged attempt should claim"); + aged.record_expired_attempt(now + MAX_RECOVERY_AGE_NANOS) + .expect("seven-day expired attempt should be recorded"); + assert_eq!(aged.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(aged.consecutive_failure_count, 1); + } + + #[test] + fn recovery_control_successor_preserves_lineage_and_generation() { + let current = control(); + let mut next = current.clone(); + let mut advanced_generation = generation(); + advanced_generation.source_schema = "rustfs-transition-transaction-v2".to_string(); + next.claim_for_source_generation("node-a", Uuid::new_v4(), 2_000_000_000, 300_000_000_000, advanced_generation) + .expect("attempt should claim"); + current + .validate_successor(&next) + .expect("a claim may adopt a newly proven source generation"); + + let mut changed = next.clone(); + let mut refreshed_generation = changed.observed_source_generation.clone(); + refreshed_generation.source_schema = "rustfs-transition-transaction-v3".to_string(); + changed + .refresh_owned_source_generation(refreshed_generation) + .expect("the current owner may refresh a newly proven source generation"); + next.validate_successor(&changed) + .expect("owned source refresh should be a legal successor"); + + let mut invalid = changed.clone(); + invalid.revision += 1; + invalid.attempt_count += 1; + assert!(matches!( + changed.validate_successor(&invalid), + Err(IlmRecoveryControlError::InvalidSuccessor(_)) + )); + } + + #[test] + fn recovery_control_view_redacts_source_and_owner_details() { + let mut control = control(); + control + .claim("secret-node-id", Uuid::new_v4(), 2_000_000_000, 300_000_000_000) + .expect("attempt should claim"); + let control_id = control.identity.source_operation_digest().expect("control id should derive"); + let view = IlmRecoveryControlView::from_control(control_id, &control).expect("view should build"); + let encoded = serde_json::to_string(&view).expect("view should encode"); + + for secret in [ + SOURCE_PATH, + "etag-a", + "etag-b", + "secret-node-id", + "12345678-90ab-cdef-1234-567890abcdef", + ] { + assert!(!encoded.contains(secret), "redacted view leaked {secret}"); + } + assert!(encoded.contains(ILM_RECOVERY_CONTROL_SCHEMA)); + assert!(encoded.contains("source_generation_sha256")); + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 82e32f598..3c866ae6f 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -23,6 +23,11 @@ use uuid::Uuid; use crate::bucket::lifecycle::config_boundary; use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE; use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; +use crate::bucket::lifecycle::recovery_control::{ + IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode, IlmRecoveryProtocol, + ObservedIlmRecoveryControl, load_recovery_control, observe_recovery_source, recovery_control_record_object_name, + save_recovery_control_if_absent, save_recovery_control_if_current, +}; use crate::bucket::lifecycle::tier_sweeper::{ delete_confirmed_transition_candidate_exact_with_lease_idempotent, delete_object_from_remote_tier_idempotent_with_manager_and_identity, @@ -44,6 +49,7 @@ const EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY: &str = "lifecycle_transit pub const DEFAULT_TRANSITION_TRANSACTION_RECOVERY_LIMIT: usize = 1_000; const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(60); const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); +const TRANSITION_RECOVERY_CONTROL_LEASE_NANOS: i64 = 15 * 60 * 1_000_000_000; pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1"; pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions"; pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix; @@ -737,6 +743,8 @@ pub enum TransitionTransactionRecoveryOutcome { RemoteCandidateDeleted, RecordDeleted, Retained, + RetainedAmbiguous(IlmRecoveryErrorCode), + OperatorRequired(IlmRecoveryErrorCode), } #[cfg(test)] @@ -817,6 +825,80 @@ async fn pause_before_transition_recovery_claim(transaction_id: Uuid) { } } +#[cfg(test)] +#[derive(Default)] +struct TransitionRecoveryTerminalBarrierState { + transaction_id: Uuid, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(test)] +pub(crate) struct TransitionRecoveryTerminalBarrier { + state: Arc, +} + +#[cfg(test)] +static TRANSITION_RECOVERY_TERMINAL_BARRIER: std::sync::OnceLock< + std::sync::Mutex>>, +> = std::sync::OnceLock::new(); + +#[cfg(test)] +impl TransitionRecoveryTerminalBarrier { + pub(crate) fn install(transaction_id: Uuid) -> Self { + let state = Arc::new(TransitionRecoveryTerminalBarrierState { + transaction_id, + ..Default::default() + }); + let mut slot = TRANSITION_RECOVERY_TERMINAL_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery terminal barrier mutex should not poison"); + assert!( + slot.is_none(), + "transition recovery terminal barrier must be installed by one test at a time" + ); + *slot = Some(Arc::clone(&state)); + drop(slot); + Self { state } + } + + pub(crate) async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("transition recovery should persist terminal control before source cleanup"); + } +} + +#[cfg(test)] +impl Drop for TransitionRecoveryTerminalBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = TRANSITION_RECOVERY_TERMINAL_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery terminal barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(test)] +async fn pause_after_transition_recovery_terminal(transaction_id: Uuid) { + let barrier = TRANSITION_RECOVERY_TERMINAL_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery terminal barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.transaction_id == transaction_id) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum TransitionOperatorProbe { @@ -1020,17 +1102,35 @@ fn transition_transaction_id_from_record_object_name(object: &str) -> Result EcstoreResult { let record_name = transition_transaction_record_object_name(observed.transaction_id).map_err(transition_transaction_store_error)?; + let now_unix_nanos = + i64::try_from(now_unix_nanos).map_err(|_| Error::other("transition transaction recovery timestamp does not fit i64"))?; + let recovery_control_identity = transition_recovery_control_identity(observed, &record_name); + let recovery_control_id = recovery_control_identity + .source_operation_digest() + .map_err(|err| Error::other(err.to_string()))?; + let control_record_name = + recovery_control_record_object_name(IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id) + .map_err(|err| Error::other(err.to_string()))?; + let control_lock = if transition_state_needs_recovery_control(observed, now_unix_nanos) { + Some( + api.new_ns_lock(RUSTFS_META_BUCKET, &format!("{control_record_name}.recovery-lock")) + .await?, + ) + } else { + None + }; + let _control_guard = match &control_lock { + Some(lock) => Some(lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?), + None => None, + }; // The synthetic key avoids nesting the recovery lock with the config // object's own I/O lock. Holding it across the bounded source proof and // remote DELETE elects one destructive recovery worker across nodes. @@ -1073,55 +1194,400 @@ async fn process_transition_transaction_record_at( return Ok(TransitionTransactionRecoveryOutcome::Retained); } - match current.state { + let mut recovery_control = if transition_state_needs_recovery_control(¤t, now_unix_nanos) { + if cleanup_terminal_transition_recovery_control( + api.clone(), + ¤t, + &record_name, + &recovery_control_identity, + &recovery_control_id, + ) + .await? + { + return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted); + } + match claim_transition_recovery_control( + api.clone(), + ¤t, + &record_name, + recovery_control_identity, + &recovery_control_id, + now_unix_nanos, + ) + .await? + { + Some(control) => Some(control), + None => return Ok(TransitionTransactionRecoveryOutcome::Retained), + } + } else { + None + }; + + let recovery = match current.state { TransitionTransactionState::Uploaded => { - if transition_transaction_ownership_is_active(¤t, now_unix_nanos) { - return Ok(TransitionTransactionRecoveryOutcome::Retained); - } - let mut cleanup = current.clone(); - cleanup - .mark_cleanup_pending( - current.fence(), - TransitionCleanupProof { - transaction_id: current.transaction_id, - write_id: current.write_id, - remote_object: current.remote_object.clone(), - remote_version: current.remote_version.clone(), - backend_fingerprint: current.backend_fingerprint, - decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit, - }, - ) - .map_err(transition_transaction_store_error)?; - #[cfg(test)] - pause_before_transition_recovery_claim(current.transaction_id).await; - match save_transition_transaction_record_if_current(api.clone(), ¤t, &cleanup).await { - Ok(()) => recover_cleanup_pending(api, &cleanup).await, - Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained), - Err(err) => Err(err), + if transition_transaction_ownership_is_active(¤t, i128::from(now_unix_nanos)) { + Ok(TransitionTransactionRecoveryOutcome::Retained) + } else { + let mut cleanup = current.clone(); + cleanup + .mark_cleanup_pending( + current.fence(), + TransitionCleanupProof { + transaction_id: current.transaction_id, + write_id: current.write_id, + remote_object: current.remote_object.clone(), + remote_version: current.remote_version.clone(), + backend_fingerprint: current.backend_fingerprint, + decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit, + }, + ) + .map_err(transition_transaction_store_error)?; + #[cfg(test)] + pause_before_transition_recovery_claim(current.transaction_id).await; + match save_transition_transaction_record_if_current(api.clone(), ¤t, &cleanup).await { + Ok(()) => recover_cleanup_pending(api.clone(), &cleanup).await, + Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => { + Ok(TransitionTransactionRecoveryOutcome::Retained) + } + Err(err) => Err(err), + } } } - TransitionTransactionState::CleanupPending => recover_cleanup_pending(api, ¤t).await, + TransitionTransactionState::CleanupPending => recover_cleanup_pending(api.clone(), ¤t).await, TransitionTransactionState::LocalCommitStarted => match local_commit_matches_transaction(api.clone(), ¤t).await { - Ok(true) => { - delete_transition_transaction_record(api, ¤t).await?; - Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) - } - Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained), - Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::Retained), + Ok(true) => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted), + Ok(false) => Ok(TransitionTransactionRecoveryOutcome::OperatorRequired( + IlmRecoveryErrorCode::LocalCommitAmbiguous, + )), + Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::OperatorRequired( + IlmRecoveryErrorCode::LocalCommitAmbiguous, + )), Err(err) => Err(err), }, TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => { - delete_transition_transaction_record(api, ¤t).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } TransitionTransactionState::UploadOutcomeUnknown => { - if transition_transaction_ownership_is_active(¤t, now_unix_nanos) { + if transition_transaction_ownership_is_active(¤t, i128::from(now_unix_nanos)) { Ok(TransitionTransactionRecoveryOutcome::Retained) } else { - recover_unknown_upload_outcome(api, ¤t).await + recover_unknown_upload_outcome(api.clone(), ¤t).await } } - TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained), + TransitionTransactionState::UploadStarted => { + if transition_transaction_ownership_is_active(¤t, i128::from(now_unix_nanos)) { + Ok(TransitionTransactionRecoveryOutcome::Retained) + } else { + Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous( + IlmRecoveryErrorCode::RemoteVersionUnknown, + )) + } + } + }; + + if let Some(mut control) = recovery_control.take() { + let source_to_delete = if matches!( + recovery, + Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted + | TransitionTransactionRecoveryOutcome::RecordDeleted) + ) { + let refreshed = + refresh_transition_recovery_control_source(api.clone(), control, &record_name, current.transaction_id).await?; + control = refreshed.0; + refreshed.1 + } else { + None + }; + persist_transition_recovery_result(api.clone(), control, &recovery, now_unix_nanos).await?; + if let Some(source) = source_to_delete { + #[cfg(test)] + pause_after_transition_recovery_terminal(source.transaction_id).await; + delete_transition_transaction_record(api, &source).await?; + } + } else if matches!( + recovery, + Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted | TransitionTransactionRecoveryOutcome::RecordDeleted) + ) { + delete_transition_transaction_record(api, ¤t).await?; + } + recovery +} + +fn transition_recovery_control_identity(transaction: &TransitionTransaction, record_name: &str) -> IlmRecoveryControlIdentity { + IlmRecoveryControlIdentity { + protocol: IlmRecoveryProtocol::TransitionTransaction, + canonical_source_path: record_name.to_string(), + stable_operation_identity: transaction.transaction_id.to_string(), + record_class: "transition_transaction_v1".to_string(), + } +} + +#[cfg(test)] +pub(crate) fn transition_recovery_control_id(transaction: &TransitionTransaction) -> Result { + let record_name = transition_transaction_record_object_name(transaction.transaction_id)?; + transition_recovery_control_identity(transaction, &record_name) + .source_operation_digest() + .map_err(|_| TransitionTransactionError::Corrupt("transition recovery control identity is invalid")) +} + +fn transition_state_needs_recovery_control(transaction: &TransitionTransaction, now_unix_nanos: i64) -> bool { + now_unix_nanos >= transaction.not_after_unix_nanos + && !matches!( + transaction.state, + TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed + ) +} + +async fn cleanup_terminal_transition_recovery_control( + api: Arc, + transaction: &TransitionTransaction, + record_name: &str, + identity: &IlmRecoveryControlIdentity, + control_id: &str, +) -> EcstoreResult { + let observed = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await { + Ok(observed) => observed, + Err(Error::ConfigNotFound) => return Ok(false), + Err(err) => return Err(err), + }; + if observed.control.classification != IlmRecoveryClassification::Terminal { + return Ok(false); + } + let source = observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await?; + let exact_source = source.is_consistent() + && source.generation == observed.control.observed_source_generation + && source.canonical_data.as_deref().is_some_and(|data| { + TransitionTransaction::decode(transaction.transaction_id, data).is_ok_and(|decoded| decoded == *transaction) + }); + if observed.control.identity != *identity || !exact_source { + return Ok(false); + } + delete_transition_transaction_record(api, transaction).await?; + Ok(true) +} + +async fn claim_transition_recovery_control( + api: Arc, + transaction: &TransitionTransaction, + record_name: &str, + identity: IlmRecoveryControlIdentity, + control_id: &str, + now_unix_nanos: i64, +) -> EcstoreResult> { + let existing = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await { + Ok(control) => Some(control), + Err(Error::ConfigNotFound) => None, + Err(err) => return Err(err), + }; + if let Some(observed) = existing.as_ref() { + if observed.control.identity != identity { + return Ok(None); + } + if observed + .control + .owner + .as_ref() + .is_some_and(|owner| owner.lease_expires_at_unix_nanos <= now_unix_nanos) + { + let mut expired = observed.control.clone(); + expired + .record_expired_attempt(now_unix_nanos) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api, observed, &expired).await?; + return Ok(None); + } + if !observed.control.should_attempt_at(now_unix_nanos) { + return Ok(None); + } + } + + let source = match observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await { + Ok(source) => source, + Err(err) => { + if let Some(observed) = existing { + persist_transition_recovery_source_failure(api, observed, now_unix_nanos).await?; + return Ok(None); + } + return Err(err); + } + }; + let source_matches = source.is_consistent() + && source.canonical_data.as_deref().is_some_and(|data| { + TransitionTransaction::decode(transaction.transaction_id, data).is_ok_and(|observed| observed == *transaction) + }); + let source_error = if source_matches { + IlmRecoveryErrorCode::None + } else if source.canonical_data.is_some() { + IlmRecoveryErrorCode::SourceGenerationChanged + } else { + IlmRecoveryErrorCode::SourceDivergent + }; + + let mut observed = match existing { + Some(control) => control, + None => { + let candidate = IlmRecoveryControl::new( + identity.clone(), + source.generation.clone(), + if source_matches { + IlmRecoveryClassification::Retrying + } else { + IlmRecoveryClassification::Corrupt + }, + now_unix_nanos, + source_error, + ) + .map_err(|err| Error::other(err.to_string()))?; + match save_recovery_control_if_absent(api.clone(), &candidate).await { + Ok(()) | Err(Error::PreconditionFailed) => {} + Err(err) => return Err(err), + } + load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await? + } + }; + if observed.control.identity != identity || !observed.control.should_attempt_at(now_unix_nanos) { + return Ok(None); + } + + let mut claimed = observed.control.clone(); + claimed + .claim_for_source_generation( + api.id.to_string(), + Uuid::new_v4(), + now_unix_nanos, + TRANSITION_RECOVERY_CONTROL_LEASE_NANOS, + source.generation, + ) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api.clone(), &observed, &claimed).await?; + observed = load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await?; + if observed.control != claimed { + return Err(Error::PreconditionFailed); + } + if !source_matches { + let mut corrupt = observed.control.clone(); + corrupt + .finish_attempt(IlmRecoveryClassification::Corrupt, source_error) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api, &observed, &corrupt).await?; + return Ok(None); + } + Ok(Some(observed)) +} + +async fn persist_transition_recovery_source_failure( + api: Arc, + observed: ObservedIlmRecoveryControl, + now_unix_nanos: i64, +) -> EcstoreResult<()> { + let mut claimed = observed.control.clone(); + claimed + .claim( + api.id.to_string(), + Uuid::new_v4(), + now_unix_nanos, + TRANSITION_RECOVERY_CONTROL_LEASE_NANOS, + ) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api.clone(), &observed, &claimed).await?; + let claimed = load_recovery_control( + api.clone(), + IlmRecoveryProtocol::TransitionTransaction, + &claimed + .identity + .source_operation_digest() + .map_err(|err| Error::other(err.to_string()))?, + ) + .await?; + let mut failed = claimed.control.clone(); + failed + .record_retryable_failure(now_unix_nanos, IlmRecoveryErrorCode::SourceUnavailable) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api, &claimed, &failed).await +} + +async fn refresh_transition_recovery_control_source( + api: Arc, + mut observed: ObservedIlmRecoveryControl, + record_name: &str, + transaction_id: Uuid, +) -> EcstoreResult<(ObservedIlmRecoveryControl, Option)> { + let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await { + Ok(transaction) => transaction, + Err(Error::ConfigNotFound) => return Ok((observed, None)), + Err(err) => return Err(err), + }; + let source = observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await?; + let exact_source = source.is_consistent() + && source + .canonical_data + .as_deref() + .is_some_and(|data| TransitionTransaction::decode(transaction_id, data).is_ok_and(|decoded| decoded == transaction)); + if !exact_source { + return Err(Error::PreconditionFailed); + } + if observed.control.observed_source_generation != source.generation { + let mut refreshed = observed.control.clone(); + refreshed + .refresh_owned_source_generation(source.generation) + .map_err(|err| Error::other(err.to_string()))?; + save_recovery_control_if_current(api.clone(), &observed, &refreshed).await?; + observed = load_recovery_control( + api, + IlmRecoveryProtocol::TransitionTransaction, + &refreshed + .identity + .source_operation_digest() + .map_err(|err| Error::other(err.to_string()))?, + ) + .await?; + if observed.control != refreshed { + return Err(Error::PreconditionFailed); + } + } + Ok((observed, Some(transaction))) +} + +async fn persist_transition_recovery_result( + api: Arc, + observed: ObservedIlmRecoveryControl, + recovery: &EcstoreResult, + now_unix_nanos: i64, +) -> EcstoreResult<()> { + let mut next = observed.control.clone(); + match recovery { + Ok( + TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted | TransitionTransactionRecoveryOutcome::RecordDeleted, + ) => next + .finish_attempt(IlmRecoveryClassification::Terminal, IlmRecoveryErrorCode::None) + .map_err(|err| Error::other(err.to_string()))?, + Ok(TransitionTransactionRecoveryOutcome::Retained) => next + .record_retryable_failure(now_unix_nanos, IlmRecoveryErrorCode::SourceGenerationChanged) + .map_err(|err| Error::other(err.to_string()))?, + Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(code)) => next + .finish_attempt(IlmRecoveryClassification::RetainedAmbiguous, *code) + .map_err(|err| Error::other(err.to_string()))?, + Ok(TransitionTransactionRecoveryOutcome::OperatorRequired(code)) => next + .finish_attempt(IlmRecoveryClassification::OperatorRequired, *code) + .map_err(|err| Error::other(err.to_string()))?, + Err(err) => next + .record_retryable_failure(now_unix_nanos, transition_recovery_error_code(err)) + .map_err(|err| Error::other(err.to_string()))?, + } + save_recovery_control_if_current(api, &observed, &next).await +} + +fn transition_recovery_error_code(err: &Error) -> IlmRecoveryErrorCode { + match err { + Error::PreconditionFailed => IlmRecoveryErrorCode::CasConflict, + Error::ConfigNotFound + | Error::FileNotFound + | Error::FileVersionNotFound + | Error::ObjectNotFound(_, _) + | Error::VersionNotFound(_, _, _) + | Error::BucketNotFound(_) => IlmRecoveryErrorCode::SourceUnavailable, + Error::SlowDown => IlmRecoveryErrorCode::BackendThrottled, + _ => IlmRecoveryErrorCode::Unknown, } } @@ -1134,10 +1600,7 @@ async fn recover_cleanup_pending( transaction: &TransitionTransaction, ) -> EcstoreResult { match local_commit_matches_transaction(api.clone(), transaction).await { - Ok(true) => { - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) - } + Ok(true) => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted), Ok(false) => delete_unreferenced_transition_candidate(api, transaction).await, Err(err) if transition_source_is_missing(&err) => delete_unreferenced_transition_candidate(api, transaction).await, Err(err) => Err(err), @@ -1157,7 +1620,6 @@ async fn delete_unreferenced_transition_candidate( return Ok(TransitionTransactionRecoveryOutcome::Retained); } delete_transition_remote_candidate(api.clone(), ¤t).await?; - delete_transition_transaction_record(api, ¤t).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } @@ -1178,24 +1640,26 @@ async fn recover_unknown_upload_outcome( .await .map_err(Error::other)? { - TransitionCandidateProbe::Missing => { - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) - } + TransitionCandidateProbe::Missing => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted), TransitionCandidateProbe::UnversionedPresent => { cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::unversioned()).await } TransitionCandidateProbe::VersionedPresent(version_id) if Uuid::parse_str(&version_id).is_ok_and(|version_id| version_id.is_nil()) => { - Ok(TransitionTransactionRecoveryOutcome::Retained) + Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous( + IlmRecoveryErrorCode::RemoteVersionUnknown, + )) } TransitionCandidateProbe::VersionedPresent(version_id) => { cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await } - TransitionCandidateProbe::Ambiguous | TransitionCandidateProbe::Unsupported => { - Ok(TransitionTransactionRecoveryOutcome::Retained) - } + TransitionCandidateProbe::Ambiguous => Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous( + IlmRecoveryErrorCode::RemoteProbeAmbiguous, + )), + TransitionCandidateProbe::Unsupported => Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous( + IlmRecoveryErrorCode::RemoteProbeUnsupported, + )), } } @@ -1323,6 +1787,11 @@ async fn recover_transition_transaction_records_with_now( false, ) .await?; + if list.is_truncated && list.next_continuation_token.is_none() { + return Err(Error::other( + "transition transaction recovery returned a truncated page without a continuation marker", + )); + } let mut stats = TransitionTransactionRecoveryStats { scanned: 0, @@ -1381,7 +1850,11 @@ async fn recover_transition_transaction_records_with_now( ) => { stats.recovered += 1; } - Ok(TransitionTransactionRecoveryOutcome::Retained) => { + Ok( + TransitionTransactionRecoveryOutcome::Retained + | TransitionTransactionRecoveryOutcome::RetainedAmbiguous(_) + | TransitionTransactionRecoveryOutcome::OperatorRequired(_), + ) => { stats.retained += 1; debug!( event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, @@ -1509,11 +1982,74 @@ fn state_requires_known_remote_version(state: TransitionTransactionState) -> boo #[cfg(test)] mod tests { use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, Ordering}; use super::*; const BACKEND_FINGERPRINT: [u8; 32] = [7; 32]; + struct RecoveryAttemptDropGuard(Arc); + + impl Drop for RecoveryAttemptDropGuard { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + async fn pending_recovery_attempt(started: Arc, dropped: Arc) -> EcstoreResult<()> { + let _drop_guard = RecoveryAttemptDropGuard(dropped); + started.notify_one(); + std::future::pending().await + } + + #[tokio::test(start_paused = true)] + async fn transition_recovery_timeout_and_cancellation_drop_inflight_attempts() { + let timeout_started = Arc::new(tokio::sync::Notify::new()); + let timeout_dropped = Arc::new(AtomicBool::new(false)); + let timeout_task = tokio::spawn({ + let started = Arc::clone(&timeout_started); + let dropped = Arc::clone(&timeout_dropped); + async move { + await_transition_transaction_recovery( + &CancellationToken::new(), + TRANSITION_TRANSACTION_RECOVERY_TIMEOUT, + pending_recovery_attempt(started, dropped), + ) + .await + } + }); + timeout_started.notified().await; + tokio::time::advance(TRANSITION_TRANSACTION_RECOVERY_TIMEOUT).await; + let timed_out = timeout_task.await.expect("timeout wrapper task should join"); + assert!(matches!(timed_out, Some(Err(_))), "outer timeout should fail the recovery pass"); + assert!(timeout_dropped.load(Ordering::SeqCst), "outer timeout must drop its in-flight attempt"); + + let cancel_token = CancellationToken::new(); + let cancel_started = Arc::new(tokio::sync::Notify::new()); + let cancel_dropped = Arc::new(AtomicBool::new(false)); + let cancel_task = tokio::spawn({ + let cancel_token = cancel_token.clone(); + let started = Arc::clone(&cancel_started); + let dropped = Arc::clone(&cancel_dropped); + async move { + await_transition_transaction_recovery( + &cancel_token, + TRANSITION_TRANSACTION_RECOVERY_TIMEOUT, + pending_recovery_attempt(started, dropped), + ) + .await + } + }); + cancel_started.notified().await; + cancel_token.cancel(); + let cancelled = cancel_task.await.expect("cancellation wrapper task should join"); + assert!(cancelled.is_none(), "outer cancellation should stop the recovery loop"); + assert!( + cancel_dropped.load(Ordering::SeqCst), + "outer cancellation must drop its in-flight attempt" + ); + } + #[derive(Default)] struct MemoryTransactionStore { records: HashMap>, @@ -1968,5 +2504,19 @@ mod tests { transition_transaction_record_object_name(Uuid::nil()), Err(TransitionTransactionError::Corrupt("transaction_id is nil")) )); + assert_eq!( + transition_transaction_id_from_record_object_name(&object).expect("canonical record path should parse"), + transaction_id + ); + for malformed in [ + object.to_ascii_uppercase(), + object.replace("/aa/aa/", "/ff/aa/"), + object.replace("/aa/aa/", "/aa/aa/extra/"), + ] { + assert!(matches!( + transition_transaction_id_from_record_object_name(&malformed), + Err(TransitionTransactionError::Corrupt(_)) + )); + } } } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 63d2b5fdd..bd597e22a 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -820,6 +820,11 @@ mod tests { manual_transition_scope_record_object_name, manual_transition_task_object_name, manual_transition_worker_result_object_name, manual_transition_worker_result_task_key, }, + recovery_control::{ + IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode, + IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, load_recovery_control, observe_recovery_source, + save_recovery_control_if_absent, + }, tier_delete_journal::{ DecommissionCheckpointTargetFailureHook, TIER_DELETE_DISPATCH_MANIFEST_PREFIX, TIER_DELETE_JOURNAL_PREFIX, TierDeleteChunkTestBarrier, TierDeleteChunkTestStage, TierDeleteDispatchBatchLimitGuard, @@ -839,12 +844,13 @@ mod tests { }, transition_transaction::{ TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError, - TransitionOperatorProbe, TransitionRecoveryClaimBarrier, TransitionRemoteVersion, TransitionSourceIdentity, - TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, - delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, - inspect_transition_transaction_for_operator, load_transition_transaction_record, - recover_transition_transaction_records, recover_transition_transaction_records_at, - save_transition_transaction_record, save_transition_transaction_record_if_current, + TransitionOperatorProbe, TransitionRecoveryClaimBarrier, TransitionRecoveryTerminalBarrier, + TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, + TransitionTransactionInit, TransitionTransactionState, delete_transition_candidate_for_operator, + finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, + load_transition_transaction_record, recover_transition_transaction_records, + recover_transition_transaction_records_at, save_transition_transaction_record, + save_transition_transaction_record_if_current, transition_recovery_control_id, transition_transaction_record_object_name, }, validate_durable_ilm_record, @@ -19234,6 +19240,105 @@ mod tests { assert!(!Arc::ptr_eq(&ctx_a, &ctx_b), "the regression requires two distinct instance contexts"); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_expires_abandoned_attempt_at_budget_bound() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store( + temp_dir.path(), + "transition-transaction-expired-attempt-budget", + &[4], + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: TransitionSourceIdentity { + bucket: "source-bucket".to_string(), + object: "source-object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1_770_000_000_000_000_000, + size: 42, + etag: "source-etag".to_string(), + version_mode: TransitionSourceVersionMode::Versioned, + }, + tier_name: "UNUSEDABANDONEDTIER".to_string(), + backend_fingerprint: [7; 32], + not_after_unix_nanos: 1, + }) + .expect("transaction should build"); + save_transition_transaction_record(store.clone(), &transaction) + .await + .expect("transaction record should persist"); + + let record_name = + transition_transaction_record_object_name(transaction.transaction_id).expect("transaction record name should derive"); + let source = observe_recovery_source( + store.clone(), + &record_name, + crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_SCHEMA, + ) + .await + .expect("transaction source generation should be observable"); + let mut control = IlmRecoveryControl::new( + IlmRecoveryControlIdentity { + protocol: IlmRecoveryProtocol::TransitionTransaction, + canonical_source_path: record_name, + stable_operation_identity: transaction.transaction_id.to_string(), + record_class: "transition_transaction_v1".to_string(), + }, + source.generation, + IlmRecoveryClassification::Retrying, + 2_000_000_000, + IlmRecoveryErrorCode::None, + ) + .expect("recovery control should build"); + let mut now = 3_000_000_000; + for _ in 1..MAX_RECOVERY_ATTEMPTS { + now = now.max(control.next_attempt_at_unix_nanos.unwrap_or(now)); + control + .claim("cancelled-or-timed-out-owner", uuid::Uuid::new_v4(), now, 1) + .expect("abandoned attempt should claim"); + control + .record_expired_attempt(now + 1) + .expect("expired attempt should consume retry budget"); + now += 2; + } + now = now.max(control.next_attempt_at_unix_nanos.expect("last retry should have a backoff")); + control + .claim("cancelled-or-timed-out-owner", uuid::Uuid::new_v4(), now, 1) + .expect("final abandoned attempt should claim"); + save_recovery_control_if_absent(store.clone(), &control) + .await + .expect("claimed recovery control should persist"); + + let stats = recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(now + 1)) + .await + .expect("recovery should account for the expired attempt"); + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0)); + + let control_id = transition_recovery_control_id(&transaction).expect("control id should derive"); + let persisted = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id) + .await + .expect("expired recovery control should remain inspectable"); + assert_eq!(persisted.control.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(persisted.control.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS)); + assert_eq!(persisted.control.consecutive_failure_count, MAX_RECOVERY_ATTEMPTS); + assert_eq!(persisted.control.last_error_code, IlmRecoveryErrorCode::AttemptLeaseExpired); + assert!(persisted.control.owner.is_none()); + assert_eq!( + transition_transaction_record_count(store).await, + 1, + "budget exhaustion must retain the source record" + ); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)] @@ -19346,6 +19451,7 @@ mod tests { ), ]; let mut expected_removes = Vec::new(); + let mut recovery_control_ids = Vec::new(); for (case, put_version, remote_version, source_mode) in cases { let mut transaction = TransitionTransaction::new(TransitionTransactionInit { deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), @@ -19383,6 +19489,8 @@ mod tests { save_transition_transaction_record(store.clone(), &transaction) .await .expect("transaction record should persist"); + recovery_control_ids + .push(transition_recovery_control_id(&transaction).expect("transition recovery control id should derive")); expected_removes.push((transaction.remote_object, put_version)); } @@ -19398,6 +19506,14 @@ mod tests { assert_eq!(actual_removes, expected_removes, "recovery must preserve each remote version shape"); assert_eq!(backend.exact_remove_count(), 2); assert_eq!(backend.object_count().await, 0); + for control_id in recovery_control_ids { + let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id) + .await + .expect("completed recovery control should remain inspectable"); + assert_eq!(control.control.classification, IlmRecoveryClassification::Terminal); + assert_eq!(control.control.attempt_count, 1); + assert!(control.control.owner.is_none()); + } let replay = recover_transition_transaction_records(store, 100, None) .await @@ -19410,6 +19526,94 @@ mod tests { ); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_resumes_source_cleanup_after_terminal_crash() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-transaction-terminal-crash", &[4])) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "TXTERMINALCRASH"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("tier lease should resolve") + .backend_identity(); + let remote_version = uuid::Uuid::new_v4().to_string(); + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: TransitionSourceIdentity { + bucket: "source-bucket".to_string(), + object: "source-object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1_770_000_000_000_000_000, + size: 42, + etag: "source-etag".to_string(), + version_mode: TransitionSourceVersionMode::Versioned, + }, + tier_name: tier_name.to_string(), + backend_fingerprint: backend_identity, + not_after_unix_nanos: 1, + }) + .expect("transaction should build"); + transaction + .advance( + transaction.fence(), + TransitionTransactionState::Uploaded, + Some(TransitionRemoteVersion::versioned(remote_version.clone())), + ) + .expect("transaction should enter uploaded state"); + backend.set_put_remote_version(Some(remote_version)).await; + let candidate = bytes::Bytes::from_static(b"terminal crash candidate"); + backend + .put( + &transaction.remote_object, + ReaderImpl::Body(candidate.clone()), + i64::try_from(candidate.len()).expect("test candidate length should fit i64"), + ) + .await + .expect("mock backend should accept candidate"); + save_transition_transaction_record(store.clone(), &transaction) + .await + .expect("transaction record should persist"); + let control_id = transition_recovery_control_id(&transaction).expect("control id should derive"); + + let barrier = TransitionRecoveryTerminalBarrier::install(transaction.transaction_id); + let recovery_store = store.clone(); + let recovery = tokio::spawn(async move { recover_transition_transaction_records(recovery_store, 100, None).await }); + barrier.wait_until_paused().await; + let terminal = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id) + .await + .expect("terminal control should persist before source cleanup"); + assert_eq!(terminal.control.classification, IlmRecoveryClassification::Terminal); + assert_eq!(transition_transaction_record_count(store.clone()).await, 1); + assert_eq!(backend.object_count().await, 0); + assert_eq!(backend.exact_remove_count(), 1); + + recovery.abort(); + assert!( + recovery + .await + .expect_err("recovery should be cancelled at the crash boundary") + .is_cancelled() + ); + drop(barrier); + + let replay = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("terminal control should resume source cleanup without another remote delete"); + assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (1, 1, 0, 0)); + assert_eq!(transition_transaction_record_count(store).await, 0); + assert_eq!(backend.exact_remove_count(), 1, "terminal replay must not repeat the remote delete"); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)] @@ -19467,6 +19671,8 @@ mod tests { save_transition_transaction_record(store.clone(), &uploaded) .await .expect("transaction record should persist"); + let recovery_control_id = + transition_recovery_control_id(&uploaded).expect("transition recovery control id should derive"); let barrier = TransitionRecoveryClaimBarrier::install(uploaded.transaction_id); let recovery_store = store.clone(); @@ -19488,11 +19694,17 @@ mod tests { .expect("recovery should treat the lost CAS as a retained transaction"); assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0)); assert_eq!( - load_transition_transaction_record(store, uploaded.transaction_id) + load_transition_transaction_record(store.clone(), uploaded.transaction_id) .await .expect("newer transaction revision must remain"), active ); + let control = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id) + .await + .expect("lost source CAS should retain a retryable recovery control"); + assert_eq!(control.control.classification, IlmRecoveryClassification::Retrying); + assert_eq!(control.control.consecutive_failure_count, 1); + assert_eq!(control.control.last_error_code, IlmRecoveryErrorCode::SourceGenerationChanged); assert_eq!(backend.object_count().await, 1, "a stale recovery must not delete the candidate"); assert_eq!(backend.remove_count().await, 0); } @@ -19794,27 +20006,15 @@ mod tests { not_after_unix_nanos: 1_780_000_000_000_000_000, }) .expect("transaction should build"); - let uploaded_fence = transaction + transaction .advance( transaction.fence(), TransitionTransactionState::Uploaded, - Some(TransitionRemoteVersion::versioned(remote_version)), + Some(TransitionRemoteVersion::versioned(remote_version.clone())), ) .expect("transaction should enter uploaded state"); - transaction - .mark_cleanup_pending( - uploaded_fence, - TransitionCleanupProof { - transaction_id: transaction.transaction_id, - write_id: transaction.write_id, - remote_object: transaction.remote_object.clone(), - remote_version: transaction.remote_version.clone(), - backend_fingerprint: transaction.backend_fingerprint, - decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit, - }, - ) - .expect("transaction should enter cleanup pending state"); let candidate = bytes::Bytes::from_static(b"cleanup pending candidate retained after failure"); + backend.set_put_remote_version(Some(remote_version)).await; backend .put( &transaction.remote_object, @@ -19826,6 +20026,8 @@ mod tests { save_transition_transaction_record(store.clone(), &transaction) .await .expect("transaction record should persist"); + let recovery_control_id = + transition_recovery_control_id(&transaction).expect("transition recovery control id should derive"); backend.set_remove_failure(true); let stats = recover_transition_transaction_records(store.clone(), 100, None) @@ -19841,6 +20043,42 @@ mod tests { assert_eq!(backend.remove_versions().await, Vec::<(String, String)>::new()); assert_eq!(backend.exact_remove_count(), 1); assert_eq!(backend.object_count().await, 1); + let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id) + .await + .expect("failed recovery control should persist"); + assert_eq!(control.control.classification, IlmRecoveryClassification::Retrying); + assert_eq!(control.control.attempt_count, 1); + assert_eq!(control.control.consecutive_failure_count, 1); + assert!( + control + .control + .next_attempt_at_unix_nanos + .is_some_and(|next| next > OffsetDateTime::now_utc().unix_timestamp_nanos() as i64) + ); + + backend.set_remove_failure(false); + let replay = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("recovery before the persisted deadline should be skipped"); + assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (1, 0, 1, 0)); + assert_eq!(backend.exact_remove_count(), 1, "persisted backoff must prevent an immediate retry"); + + let retry_at = control + .control + .next_attempt_at_unix_nanos + .expect("retry deadline should persist"); + let retried = recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(retry_at) + 1) + .await + .expect("recovery at the persisted deadline should retry the advanced source generation"); + assert_eq!((retried.scanned, retried.recovered, retried.retained, retried.failed), (1, 1, 0, 0)); + assert_eq!(backend.exact_remove_count(), 2); + assert_eq!(backend.object_count().await, 0); + assert_eq!(transition_transaction_record_count(store.clone()).await, 0); + let terminal = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id) + .await + .expect("completed retry control should remain inspectable"); + assert_eq!(terminal.control.classification, IlmRecoveryClassification::Terminal); + assert_eq!(terminal.control.attempt_count, 2); } #[cfg(feature = "test-util")] @@ -20141,6 +20379,10 @@ mod tests { local_commit_started .advance(local_commit_started.fence(), TransitionTransactionState::LocalCommitStarted, None) .expect("transaction should enter local commit state"); + let upload_started_control_id = + transition_recovery_control_id(&upload_started).expect("upload-started control id should derive"); + let local_commit_control_id = + transition_recovery_control_id(&local_commit_started).expect("local-commit control id should derive"); backend.set_put_remote_version(Some(remote_version)).await; for transaction in [&upload_started, &local_commit_started] { @@ -20171,6 +20413,19 @@ mod tests { assert_eq!(backend.object_count().await, 2, "recovery must not delete an unproven remote candidate"); assert_eq!(backend.remove_count().await, 0); assert_eq!(backend.exact_remove_count(), 0); + let upload_started_control = + load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &upload_started_control_id) + .await + .expect("upload-started control should persist"); + assert_eq!( + upload_started_control.control.classification, + IlmRecoveryClassification::RetainedAmbiguous + ); + let local_commit_control = + load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id) + .await + .expect("local-commit control should persist"); + assert_eq!(local_commit_control.control.classification, IlmRecoveryClassification::OperatorRequired); } #[cfg(feature = "test-util")] @@ -20521,6 +20776,12 @@ mod tests { "an unsupported provider probe must retain the unknown upload" ); assert_eq!(transition_transaction_record_count(store.clone()).await, 1); + let recovery_control_id = + transition_recovery_control_id(&transaction).expect("transition recovery control id should derive"); + let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id) + .await + .expect("unsupported probe control should persist"); + assert_eq!(control.control.classification, IlmRecoveryClassification::RetainedAmbiguous); assert!( backend.contains(&transaction.remote_object).await, "unsupported recovery must not delete the candidate" diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index c418af8b3..74788d0d9 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -18,18 +18,20 @@ use crate::admin::runtime_sources::object_store_from_extensions; use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; use crate::admin::storage_api::error::StorageError; use crate::admin::storage_api::lifecycle::{ - ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink, - ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission, - ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError, - claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current, - delete_transition_candidate_for_operator, enqueue_transition_for_existing_objects_scoped, - finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, + IlmRecoveryClassification, IlmRecoveryProtocol, ManualTransitionCancelCheck, ManualTransitionJobRecord, + ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot, ManualTransitionRunOptions, + ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, + TransitionOperatorDeleteResult, TransitionOperatorError, claim_manual_transition_scope_admission, + delete_manual_transition_scope_admission_if_current, delete_transition_candidate_for_operator, + enqueue_transition_for_existing_objects_scoped, finalize_missing_transition_transaction_for_operator, + inspect_recovery_control, inspect_transition_transaction_for_operator, list_recovery_controls, load_manual_transition_job_record, load_manual_transition_scope_admission, manual_transition_job_lease_expired, manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired, persist_manual_transition_job_progress_if_owned, renew_manual_transition_job_lease_if_owned, request_manual_transition_job_cancel, save_manual_transition_job_record, update_manual_transition_job_record, }; use crate::admin::storage_api::runtime::ECStore; +use crate::admin::storage_api::s3::{S3ErrorCode as AdminS3ErrorCode, error as admin_s3_error}; use crate::admin::utils::json_response; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use http::HeaderMap; @@ -230,9 +232,48 @@ pub fn register_ilm_transition_route(r: &mut S3Router) -> std::i format!("{ADMIN_PREFIX}/v3/ilm/transition/reconcile/{{transaction_id}}").as_str(), AdminOperation(&TransitionReconcileApplyHandler {}), )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/ilm/recovery/records").as_str(), + AdminOperation(&IlmRecoveryControlListHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/ilm/recovery/records/{{control_id}}").as_str(), + AdminOperation(&IlmRecoveryControlInspectHandler {}), + )?; Ok(()) } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct IlmRecoveryControlListQuery { + protocol: IlmRecoveryProtocol, + #[serde(default)] + classification: Option, + #[serde(default = "default_recovery_control_list_limit")] + limit: usize, + #[serde(default)] + marker: Option, +} + +const fn default_recovery_control_list_limit() -> usize { + 100 +} + +fn parse_recovery_control_list_query(query: Option<&str>) -> S3Result { + let query = query.ok_or_else(|| admin_s3_error(AdminS3ErrorCode::InvalidRequest, "protocol is required"))?; + let parsed: IlmRecoveryControlListQuery = serde_urlencoded::from_bytes(query.as_bytes()) + .map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control query"))?; + if !(1..=1_000).contains(&parsed.limit) { + return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "limit must be between 1 and 1000")); + } + if parsed.marker.as_ref().is_some_and(|marker| marker.is_empty()) { + return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "marker must not be empty")); + } + Ok(parsed) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ManualTransitionRunMode { EnqueueOnly, @@ -423,6 +464,26 @@ 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 + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control id")); + } + Ok(control_id.to_string()) +} + +fn map_recovery_control_error(err: StorageError) -> S3Error { + if err == StorageError::ConfigNotFound { + admin_s3_error(AdminS3ErrorCode::NoSuchKey, "ILM recovery control not found") + } else { + admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery control request failed") + } +} + fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error { match err { TransitionOperatorError::NotFound => s3_error!(NoSuchKey, "transition transaction not found"), @@ -1031,6 +1092,40 @@ impl Operation for TransitionReconcileInspectHandler { } } +pub struct IlmRecoveryControlListHandler {} + +#[async_trait::async_trait] +impl Operation for IlmRecoveryControlListHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?; + let query = parse_recovery_control_list_query(req.uri.query())?; + let Some(store) = object_store_from_extensions(&req.extensions) else { + return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized")); + }; + let page = list_recovery_controls(store, query.protocol, query.classification, query.limit, query.marker) + .await + .map_err(map_recovery_control_error)?; + json_response(StatusCode::OK, &page) + } +} + +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 control_id = recovery_control_id_from_params(¶ms)?; + let Some(store) = object_store_from_extensions(&req.extensions) else { + return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized")); + }; + let control = inspect_recovery_control(store, &control_id) + .await + .map_err(map_recovery_control_error)?; + json_response(StatusCode::OK, &control) + } +} + pub struct TransitionReconcileApplyHandler {} #[async_trait::async_trait] @@ -1104,6 +1199,49 @@ mod tests { f(&matched.params) } + fn with_recovery_control_params(path: &str, f: impl FnOnce(&Params<'_, '_>) -> T) -> T { + let mut router = Router::new(); + router + .insert("/rustfs/admin/v3/ilm/recovery/records/{control_id}", ()) + .expect("route should insert"); + let matched = router.at(path).expect("route should match"); + f(&matched.params) + } + + #[test] + fn recovery_control_query_is_bounded_and_strict() { + let query = parse_recovery_control_list_query(Some("protocol=transition_transaction")) + .expect("minimal recovery query should parse"); + assert_eq!(query.protocol, IlmRecoveryProtocol::TransitionTransaction); + assert_eq!(query.classification, None); + assert_eq!(query.limit, 100); + + let filtered = parse_recovery_control_list_query(Some( + "protocol=tier_delete_journal&classification=retained_ambiguous&limit=1000&marker=opaque", + )) + .expect("bounded filtered query should parse"); + assert_eq!(filtered.protocol, IlmRecoveryProtocol::TierDeleteJournal); + assert_eq!(filtered.classification, Some(IlmRecoveryClassification::RetainedAmbiguous)); + assert_eq!(filtered.limit, 1000); + assert!(parse_recovery_control_list_query(None).is_err()); + assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&limit=0")).is_err()); + assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&limit=1001")).is_err()); + assert!(parse_recovery_control_list_query(Some("protocol=unknown")).is_err()); + assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&extra=true")).is_err()); + } + + #[test] + fn recovery_control_id_is_canonical_lowercase_sha256() { + let id = "ab".repeat(32); + with_recovery_control_params(&format!("/rustfs/admin/v3/ilm/recovery/records/{id}"), |params| { + assert_eq!(recovery_control_id_from_params(params).expect("control id should parse"), id); + }); + let uppercase = "AB".repeat(32); + with_recovery_control_params(&format!("/rustfs/admin/v3/ilm/recovery/records/{uppercase}"), |params| { + assert!(recovery_control_id_from_params(params).is_err()) + }); + } + fn manual_transition_job_request(method: Method, path: &'static str) -> S3Request { S3Request { input: Body::empty(), diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 03adac373..bbe1e9a11 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -232,6 +232,9 @@ pub(crate) mod lifecycle { pub(crate) type ManualTransitionRunOptions = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunOptions; pub(crate) type ManualTransitionRunReport = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunReport; + pub(crate) use super::ecstore_bucket::lifecycle::recovery_control::{ + IlmRecoveryClassification, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls, + }; pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{ TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator,