mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-08 04:58:12 +00:00
feat(ilm): retry retained transition recovery (#7308)
This commit is contained in:
cxymds
@@ -93,8 +93,9 @@ pub mod bucket {
|
||||
pub mod transition_transaction {
|
||||
pub use crate::bucket::lifecycle::transition_transaction::{
|
||||
TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,
|
||||
delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
|
||||
inspect_transition_transaction_for_operator,
|
||||
TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus, delete_transition_candidate_for_operator,
|
||||
finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator,
|
||||
inspect_transition_transaction_for_operator, retry_transition_recovery_for_operator,
|
||||
};
|
||||
#[cfg(feature = "test-util")]
|
||||
pub use crate::bucket::lifecycle::transition_transaction::{
|
||||
|
||||
@@ -499,6 +499,26 @@ impl IlmRecoveryControl {
|
||||
self.validate()
|
||||
}
|
||||
|
||||
pub fn retry_for_operator(&mut self, expected_source_generation: &IlmRecoverySourceGeneration) -> Result<()> {
|
||||
if self.owner.is_some()
|
||||
|| !matches!(
|
||||
self.classification,
|
||||
IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired
|
||||
)
|
||||
|| self.attempt_count == u64::MAX
|
||||
|| &self.observed_source_generation != expected_source_generation
|
||||
{
|
||||
return Err(IlmRecoveryControlError::InvalidSuccessor(
|
||||
"operator retry requires the exact ownerless retained source generation",
|
||||
));
|
||||
}
|
||||
self.bump_revision()?;
|
||||
self.classification = IlmRecoveryClassification::Retrying;
|
||||
self.consecutive_failure_count = 0;
|
||||
self.next_attempt_at_unix_nanos = None;
|
||||
self.validate()
|
||||
}
|
||||
|
||||
pub fn validate_successor(&self, next: &Self) -> Result<()> {
|
||||
self.validate()?;
|
||||
next.validate()?;
|
||||
@@ -528,6 +548,14 @@ impl IlmRecoveryControl {
|
||||
{
|
||||
self.validate_operator_abandon_successor(next)
|
||||
}
|
||||
(None, None)
|
||||
if matches!(
|
||||
self.classification,
|
||||
IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired
|
||||
) && next.classification == IlmRecoveryClassification::Retrying =>
|
||||
{
|
||||
self.validate_operator_retry_successor(next)
|
||||
}
|
||||
(None, None) => Err(IlmRecoveryControlError::InvalidSuccessor(
|
||||
"ownerless control cannot advance without a claim",
|
||||
)),
|
||||
@@ -550,6 +578,22 @@ impl IlmRecoveryControl {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_operator_retry_successor(&self, next: &Self) -> Result<()> {
|
||||
if next.observed_source_generation != self.observed_source_generation
|
||||
|| next.attempt_count != self.attempt_count
|
||||
|| 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()
|
||||
|| next.last_error_code != self.last_error_code
|
||||
{
|
||||
return Err(IlmRecoveryControlError::InvalidSuccessor(
|
||||
"operator retry changed recovery history or source generation",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_claim_successor(&self, next: &Self) -> Result<()> {
|
||||
if self.classification != IlmRecoveryClassification::Retrying
|
||||
|| next.classification != IlmRecoveryClassification::Retrying
|
||||
@@ -1347,6 +1391,69 @@ mod tests {
|
||||
assert!(previous.validate_successor(&mutated_history).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn operator_retry_rearms_exact_retained_generation_without_resetting_history() {
|
||||
for classification in [
|
||||
IlmRecoveryClassification::RetainedAmbiguous,
|
||||
IlmRecoveryClassification::OperatorRequired,
|
||||
] {
|
||||
let mut retained = control();
|
||||
retained
|
||||
.claim("node-a", Uuid::new_v4(), 2_000_000_000, 1)
|
||||
.expect("attempt should claim");
|
||||
retained
|
||||
.record_retryable_failure(2_000_000_001, IlmRecoveryErrorCode::BackendTimeout)
|
||||
.expect("failure should persist");
|
||||
retained.classification = classification;
|
||||
retained.next_attempt_at_unix_nanos = None;
|
||||
if classification == IlmRecoveryClassification::OperatorRequired {
|
||||
retained.attempt_count = u64::from(MAX_RECOVERY_ATTEMPTS);
|
||||
retained.consecutive_failure_count = MAX_RECOVERY_ATTEMPTS;
|
||||
}
|
||||
retained.validate().expect("retained control should remain valid");
|
||||
|
||||
let previous = retained.clone();
|
||||
retained
|
||||
.retry_for_operator(&previous.observed_source_generation)
|
||||
.expect("exact retained generation should be retryable");
|
||||
previous
|
||||
.validate_successor(&retained)
|
||||
.expect("operator retry should be a valid successor");
|
||||
assert_eq!(retained.classification, IlmRecoveryClassification::Retrying);
|
||||
assert_eq!(retained.revision, previous.revision + 1);
|
||||
assert_eq!(retained.attempt_count, previous.attempt_count);
|
||||
assert_eq!(retained.first_failure_at_unix_nanos, previous.first_failure_at_unix_nanos);
|
||||
assert_eq!(retained.last_failure_at_unix_nanos, previous.last_failure_at_unix_nanos);
|
||||
assert_eq!(retained.last_error_code, previous.last_error_code);
|
||||
assert_eq!(retained.consecutive_failure_count, 0);
|
||||
assert_eq!(retained.next_attempt_at_unix_nanos, None);
|
||||
assert!(retained.should_attempt_at(2_000_000_002));
|
||||
|
||||
if classification == IlmRecoveryClassification::OperatorRequired {
|
||||
retained
|
||||
.claim("node-b", Uuid::new_v4(), 2_000_000_002, 1)
|
||||
.expect("operator retry should authorize one new bounded attempt");
|
||||
retained
|
||||
.record_retryable_failure(2_000_000_003, IlmRecoveryErrorCode::BackendTimeout)
|
||||
.expect("the bounded attempt failure should persist");
|
||||
assert_eq!(retained.classification, IlmRecoveryClassification::OperatorRequired);
|
||||
assert_eq!(retained.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS) + 1);
|
||||
assert_eq!(retained.consecutive_failure_count, 1);
|
||||
}
|
||||
|
||||
let mut wrong_generation = previous;
|
||||
let mut changed_generation = wrong_generation.observed_source_generation.clone();
|
||||
changed_generation.source_etag = "changed".to_string();
|
||||
assert!(wrong_generation.retry_for_operator(&changed_generation).is_err());
|
||||
}
|
||||
|
||||
let mut exhausted = control();
|
||||
exhausted.classification = IlmRecoveryClassification::OperatorRequired;
|
||||
exhausted.attempt_count = u64::MAX;
|
||||
let generation = exhausted.observed_source_generation.clone();
|
||||
assert!(exhausted.retry_for_operator(&generation).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_control_view_redacts_source_and_owner_details() {
|
||||
let mut control = control();
|
||||
|
||||
@@ -34,7 +34,7 @@ use crate::bucket::lifecycle::tier_sweeper::{
|
||||
};
|
||||
use crate::disk::RUSTFS_META_BUCKET;
|
||||
use crate::error::{Error, Result as EcstoreResult};
|
||||
use crate::object_api::ObjectOptions;
|
||||
use crate::object_api::{ObjectInfo, ObjectOptions};
|
||||
use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe};
|
||||
use crate::storage_api_contracts::{
|
||||
list::ListOperations as _,
|
||||
@@ -954,6 +954,10 @@ pub enum TransitionOperatorError {
|
||||
expected: String,
|
||||
actual: TransitionOperatorProbe,
|
||||
},
|
||||
#[error("transition recovery control is stale")]
|
||||
StaleRecoveryControl,
|
||||
#[error("transition recovery control is not eligible for operator retry")]
|
||||
RetryNotAllowed,
|
||||
#[error("transition transaction store failed: {0}")]
|
||||
Store(#[source] Error),
|
||||
#[error("remote tier reconciliation failed: {0}")]
|
||||
@@ -962,6 +966,179 @@ pub enum TransitionOperatorError {
|
||||
|
||||
type TransitionOperatorResult<T> = std::result::Result<T, TransitionOperatorError>;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct TransitionRecoveryRetryStatus {
|
||||
pub control_id: String,
|
||||
pub transaction_id: Uuid,
|
||||
pub state: TransitionTransactionState,
|
||||
pub classification: IlmRecoveryClassification,
|
||||
pub control_revision: u64,
|
||||
pub attempt_count: u64,
|
||||
pub consecutive_failure_count: u32,
|
||||
pub last_error_code: IlmRecoveryErrorCode,
|
||||
pub source_generation_sha256: String,
|
||||
pub copy_set_sha256: String,
|
||||
pub retry_ready: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub retry_not_ready_reason: Option<&'static str>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct TransitionRecoveryRetryResult {
|
||||
pub control_id: String,
|
||||
pub transaction_id: Uuid,
|
||||
pub previous_revision: u64,
|
||||
pub revision: u64,
|
||||
pub classification: IlmRecoveryClassification,
|
||||
pub attempt_count: u64,
|
||||
pub source_generation_sha256: String,
|
||||
}
|
||||
|
||||
struct TransitionRecoveryRetryContext {
|
||||
observed: ObservedIlmRecoveryControl,
|
||||
transaction: TransitionTransaction,
|
||||
source_generation_sha256: String,
|
||||
}
|
||||
|
||||
fn transition_recovery_retry_readiness(control: &IlmRecoveryControl) -> (bool, Option<&'static str>) {
|
||||
if control.owner.is_some() {
|
||||
return (false, Some("attempt_owned"));
|
||||
}
|
||||
match control.classification {
|
||||
IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired => (true, None),
|
||||
IlmRecoveryClassification::Retrying => (false, Some("already_retrying")),
|
||||
IlmRecoveryClassification::Corrupt => (false, Some("source_corrupt")),
|
||||
IlmRecoveryClassification::Abandoned => (false, Some("source_abandoned")),
|
||||
IlmRecoveryClassification::Terminal => (false, Some("source_terminal")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_transition_recovery_retry_context(
|
||||
api: Arc<ECStore>,
|
||||
control_id: &str,
|
||||
) -> TransitionOperatorResult<TransitionRecoveryRetryContext> {
|
||||
let observed = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await {
|
||||
Ok(observed) => observed,
|
||||
Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound),
|
||||
Err(err) => return Err(TransitionOperatorError::Store(err)),
|
||||
};
|
||||
let transaction_id = Uuid::parse_str(&observed.control.identity.stable_operation_identity)
|
||||
.ok()
|
||||
.filter(|transaction_id| !transaction_id.is_nil())
|
||||
.ok_or(TransitionOperatorError::StaleRecoveryControl)?;
|
||||
let canonical_path = transition_transaction_record_object_name(transaction_id)
|
||||
.map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
|
||||
if observed.control.identity.canonical_source_path != canonical_path
|
||||
|| observed.control.identity.record_class != "transition_transaction_v1"
|
||||
{
|
||||
return Err(TransitionOperatorError::StaleRecoveryControl);
|
||||
}
|
||||
let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
|
||||
Ok(transaction) => transaction,
|
||||
Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound),
|
||||
Err(err) => return Err(TransitionOperatorError::Store(err)),
|
||||
};
|
||||
let source = observe_recovery_source(api, &canonical_path, TRANSITION_TRANSACTION_SCHEMA)
|
||||
.await
|
||||
.map_err(TransitionOperatorError::Store)?;
|
||||
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_id, data).is_ok_and(|decoded| decoded == transaction));
|
||||
if !exact_source {
|
||||
return Err(TransitionOperatorError::StaleRecoveryControl);
|
||||
}
|
||||
let generation = serde_json::to_vec(&observed.control.observed_source_generation)
|
||||
.map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
|
||||
Ok(TransitionRecoveryRetryContext {
|
||||
observed,
|
||||
transaction,
|
||||
source_generation_sha256: hex_sha256(&generation, ToOwned::to_owned),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn inspect_transition_recovery_retry_for_operator(
|
||||
api: Arc<ECStore>,
|
||||
control_id: &str,
|
||||
) -> TransitionOperatorResult<TransitionRecoveryRetryStatus> {
|
||||
let context = load_transition_recovery_retry_context(api, control_id).await?;
|
||||
let (retry_ready, retry_not_ready_reason) = transition_recovery_retry_readiness(&context.observed.control);
|
||||
Ok(TransitionRecoveryRetryStatus {
|
||||
control_id: control_id.to_string(),
|
||||
transaction_id: context.transaction.transaction_id,
|
||||
state: context.transaction.state,
|
||||
classification: context.observed.control.classification,
|
||||
control_revision: context.observed.control.revision,
|
||||
attempt_count: context.observed.control.attempt_count,
|
||||
consecutive_failure_count: context.observed.control.consecutive_failure_count,
|
||||
last_error_code: context.observed.control.last_error_code,
|
||||
source_generation_sha256: context.source_generation_sha256,
|
||||
copy_set_sha256: context.observed.control.observed_source_generation.copy_set_sha256.clone(),
|
||||
retry_ready,
|
||||
retry_not_ready_reason,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn retry_transition_recovery_for_operator(
|
||||
api: Arc<ECStore>,
|
||||
control_id: &str,
|
||||
expected_control_revision: u64,
|
||||
expected_source_generation_sha256: &str,
|
||||
) -> TransitionOperatorResult<TransitionRecoveryRetryResult> {
|
||||
let control_object = recovery_control_record_object_name(IlmRecoveryProtocol::TransitionTransaction, control_id)
|
||||
.map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
|
||||
let retry_lock = api
|
||||
.new_ns_lock(RUSTFS_META_BUCKET, &format!("{control_object}.recovery-lock"))
|
||||
.await
|
||||
.map_err(TransitionOperatorError::Store)?;
|
||||
let retry_guard = retry_lock
|
||||
.get_write_lock(crate::set_disk::get_lock_acquire_timeout())
|
||||
.await
|
||||
.map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
|
||||
let context = load_transition_recovery_retry_context(api.clone(), control_id).await?;
|
||||
let (retry_ready, _) = transition_recovery_retry_readiness(&context.observed.control);
|
||||
if !retry_ready {
|
||||
return Err(TransitionOperatorError::RetryNotAllowed);
|
||||
}
|
||||
if retry_guard.is_lock_lost()
|
||||
|| expected_control_revision == 0
|
||||
|| context.observed.control.revision != expected_control_revision
|
||||
|| context.source_generation_sha256 != expected_source_generation_sha256
|
||||
{
|
||||
return Err(TransitionOperatorError::StaleRecoveryControl);
|
||||
}
|
||||
let previous_revision = context.observed.control.revision;
|
||||
let mut next = context.observed.control.clone();
|
||||
next.retry_for_operator(&context.observed.control.observed_source_generation)
|
||||
.map_err(|_| TransitionOperatorError::RetryNotAllowed)?;
|
||||
if retry_guard.is_lock_lost() {
|
||||
return Err(TransitionOperatorError::StaleRecoveryControl);
|
||||
}
|
||||
save_recovery_control_if_current(api.clone(), &context.observed, &next)
|
||||
.await
|
||||
.map_err(|err| match err {
|
||||
Error::PreconditionFailed => TransitionOperatorError::StaleRecoveryControl,
|
||||
err => TransitionOperatorError::Store(err),
|
||||
})?;
|
||||
let persisted = load_recovery_control(api, IlmRecoveryProtocol::TransitionTransaction, control_id)
|
||||
.await
|
||||
.map_err(TransitionOperatorError::Store)?;
|
||||
if retry_guard.is_lock_lost() || persisted.control != next {
|
||||
return Err(TransitionOperatorError::StaleRecoveryControl);
|
||||
}
|
||||
Ok(TransitionRecoveryRetryResult {
|
||||
control_id: control_id.to_string(),
|
||||
transaction_id: context.transaction.transaction_id,
|
||||
previous_revision,
|
||||
revision: persisted.control.revision,
|
||||
classification: persisted.control.classification,
|
||||
attempt_count: persisted.control.attempt_count,
|
||||
source_generation_sha256: context.source_generation_sha256,
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_operator_reconcile_transaction(
|
||||
transaction: &TransitionTransaction,
|
||||
now_unix_nanos: i128,
|
||||
@@ -1706,12 +1883,27 @@ async fn local_commit_matches_transaction(api: Arc<ECStore>, transaction: &Trans
|
||||
.get_object_info(&transaction.source.bucket, &transaction.source.object, &opts)
|
||||
.await?;
|
||||
let transitioned = &object.transitioned_object;
|
||||
Ok(transitioned.status == TRANSITION_COMPLETE
|
||||
Ok(local_object_matches_transition_source(&object, &transaction.source)
|
||||
&& transitioned.status == TRANSITION_COMPLETE
|
||||
&& transitioned.name == transaction.remote_object
|
||||
&& transitioned.tier == transaction.tier_name
|
||||
&& transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_default())
|
||||
}
|
||||
|
||||
fn local_object_matches_transition_source(object: &ObjectInfo, source: &TransitionSourceIdentity) -> bool {
|
||||
let observed_version_id = object.version_id.filter(|version_id| !version_id.is_nil());
|
||||
let observed_mod_time = object
|
||||
.mod_time
|
||||
.and_then(|mod_time| i64::try_from(mod_time.unix_timestamp_nanos()).ok());
|
||||
object.bucket == source.bucket
|
||||
&& object.name == source.object
|
||||
&& observed_version_id == source.version_id
|
||||
&& object.data_dir == Some(source.data_dir)
|
||||
&& observed_mod_time == Some(source.mod_time_unix_nanos)
|
||||
&& object.size == source.size
|
||||
&& object.etag.as_deref() == Some(source.etag.as_str())
|
||||
}
|
||||
|
||||
fn transition_source_lookup_options(transaction: &TransitionTransaction) -> ObjectOptions {
|
||||
ObjectOptions {
|
||||
version_id: match transaction.source.version_mode {
|
||||
@@ -2183,6 +2375,41 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_commit_proof_requires_the_complete_source_identity() {
|
||||
let source = source_identity(TransitionSourceVersionMode::Versioned);
|
||||
let exact = ObjectInfo {
|
||||
bucket: source.bucket.clone(),
|
||||
name: source.object.clone(),
|
||||
version_id: source.version_id,
|
||||
data_dir: Some(source.data_dir),
|
||||
mod_time: Some(
|
||||
time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(source.mod_time_unix_nanos))
|
||||
.expect("source timestamp should be valid"),
|
||||
),
|
||||
size: source.size,
|
||||
etag: Some(source.etag.clone()),
|
||||
..Default::default()
|
||||
};
|
||||
assert!(local_object_matches_transition_source(&exact, &source));
|
||||
|
||||
let mut changed = exact.clone();
|
||||
changed.version_id = Some(Uuid::new_v4());
|
||||
assert!(!local_object_matches_transition_source(&changed, &source));
|
||||
changed = exact.clone();
|
||||
changed.data_dir = Some(Uuid::new_v4());
|
||||
assert!(!local_object_matches_transition_source(&changed, &source));
|
||||
changed = exact.clone();
|
||||
changed.mod_time = changed.mod_time.map(|value| value + Duration::from_nanos(1));
|
||||
assert!(!local_object_matches_transition_source(&changed, &source));
|
||||
changed = exact.clone();
|
||||
changed.size += 1;
|
||||
assert!(!local_object_matches_transition_source(&changed, &source));
|
||||
changed = exact;
|
||||
changed.etag = Some("different-etag".to_string());
|
||||
assert!(!local_object_matches_transition_source(&changed, &source));
|
||||
}
|
||||
|
||||
fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
|
||||
TransitionCleanupProof {
|
||||
transaction_id: transaction.transaction_id,
|
||||
|
||||
@@ -862,9 +862,10 @@ mod tests {
|
||||
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,
|
||||
finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator,
|
||||
inspect_transition_transaction_for_operator, load_transition_transaction_record,
|
||||
recover_transition_transaction_records, recover_transition_transaction_records_at,
|
||||
retry_transition_recovery_for_operator, save_transition_transaction_record,
|
||||
save_transition_transaction_record_if_current, transition_recovery_control_id,
|
||||
transition_transaction_record_object_name,
|
||||
},
|
||||
@@ -20629,7 +20630,7 @@ mod tests {
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
version_id: None,
|
||||
data_dir: uuid::Uuid::new_v4(),
|
||||
data_dir: original.data_dir.expect("source object should have data_dir"),
|
||||
mod_time_unix_nanos: original
|
||||
.mod_time
|
||||
.expect("source object should have mod_time")
|
||||
@@ -20763,7 +20764,7 @@ mod tests {
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
version_id: None,
|
||||
data_dir: uuid::Uuid::new_v4(),
|
||||
data_dir: original.data_dir.expect("source object should have data_dir"),
|
||||
mod_time_unix_nanos: original
|
||||
.mod_time
|
||||
.expect("source object should have mod_time")
|
||||
@@ -20904,10 +20905,77 @@ mod tests {
|
||||
IlmRecoveryClassification::RetainedAmbiguous
|
||||
);
|
||||
let local_commit_control =
|
||||
load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
|
||||
load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
|
||||
.await
|
||||
.expect("local-commit control should persist");
|
||||
assert_eq!(local_commit_control.control.classification, IlmRecoveryClassification::OperatorRequired);
|
||||
|
||||
let upload_status = inspect_transition_recovery_retry_for_operator(store.clone(), &upload_started_control_id)
|
||||
.await
|
||||
.expect("retained upload should be inspectable for a bounded retry");
|
||||
let local_status = inspect_transition_recovery_retry_for_operator(store.clone(), &local_commit_control_id)
|
||||
.await
|
||||
.expect("operator-required local commit should be inspectable for a bounded retry");
|
||||
assert!(upload_status.retry_ready);
|
||||
assert!(local_status.retry_ready);
|
||||
assert!(matches!(
|
||||
retry_transition_recovery_for_operator(
|
||||
store.clone(),
|
||||
&upload_started_control_id,
|
||||
upload_status.control_revision + 1,
|
||||
&upload_status.source_generation_sha256,
|
||||
)
|
||||
.await,
|
||||
Err(TransitionOperatorError::StaleRecoveryControl)
|
||||
));
|
||||
|
||||
let put_count_before_retry = backend.put_count().await;
|
||||
let get_count_before_retry = backend.get_count().await;
|
||||
let remove_count_before_retry = backend.remove_count().await;
|
||||
let upload_retry = retry_transition_recovery_for_operator(
|
||||
store.clone(),
|
||||
&upload_started_control_id,
|
||||
upload_status.control_revision,
|
||||
&upload_status.source_generation_sha256,
|
||||
)
|
||||
.await
|
||||
.expect("exact retained upload generation should be rearmed");
|
||||
let local_retry = retry_transition_recovery_for_operator(
|
||||
store.clone(),
|
||||
&local_commit_control_id,
|
||||
local_status.control_revision,
|
||||
&local_status.source_generation_sha256,
|
||||
)
|
||||
.await
|
||||
.expect("exact operator-required local commit generation should be rearmed");
|
||||
assert_eq!(upload_retry.classification, IlmRecoveryClassification::Retrying);
|
||||
assert_eq!(local_retry.classification, IlmRecoveryClassification::Retrying);
|
||||
assert_eq!(upload_retry.attempt_count, upload_status.attempt_count);
|
||||
assert_eq!(local_retry.attempt_count, local_status.attempt_count);
|
||||
assert_eq!(backend.put_count().await, put_count_before_retry);
|
||||
assert_eq!(backend.get_count().await, get_count_before_retry);
|
||||
assert_eq!(backend.remove_count().await, remove_count_before_retry);
|
||||
assert_eq!(backend.exact_remove_count(), 0, "operator retry must not directly issue remote DELETE");
|
||||
|
||||
let retried = recover_transition_transaction_records(store.clone(), 100, None)
|
||||
.await
|
||||
.expect("rearmed records should be re-evaluated through normal recovery");
|
||||
assert_eq!((retried.scanned, retried.recovered, retried.retained, retried.failed), (2, 0, 2, 0));
|
||||
let upload_retained =
|
||||
load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &upload_started_control_id)
|
||||
.await
|
||||
.expect("upload retry result should persist");
|
||||
let local_retained = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
|
||||
.await
|
||||
.expect("local commit retry result should persist");
|
||||
assert_eq!(upload_retained.control.classification, IlmRecoveryClassification::RetainedAmbiguous);
|
||||
assert_eq!(local_retained.control.classification, IlmRecoveryClassification::OperatorRequired);
|
||||
assert_eq!(upload_retained.control.attempt_count, upload_status.attempt_count + 1);
|
||||
assert_eq!(local_retained.control.attempt_count, local_status.attempt_count + 1);
|
||||
assert_eq!(backend.put_count().await, put_count_before_retry);
|
||||
assert_eq!(backend.get_count().await, get_count_before_retry);
|
||||
assert_eq!(backend.remove_count().await, remove_count_before_retry);
|
||||
assert_eq!(backend.exact_remove_count(), 0);
|
||||
}
|
||||
|
||||
#[cfg(feature = "test-util")]
|
||||
|
||||
Reference in New Issue
Block a user