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

fix: fence transition transaction recovery (#7095)

Browse Source
This commit is contained in:
cxymds
2026-09-03 18:38:30 +08:00
committed by GitHub
parent 703766677b
commit a6cb34c7a4
8 changed files with 984 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/api/mod.rs
+1
View File
@@ -78,6 +78,7 @@ pub mod bucket {
#[cfg(feature = "test-util")]
pub use crate::bucket::lifecycle::transition_transaction::{
TransitionTransactionRecoveryStats, recover_transition_transaction_records,
recover_transition_transaction_records_at,
};
}
crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs
+1
View File
@@ -575,6 +575,7 @@ pub(crate) async fn delete_confirmed_transition_candidate_exact_with_lease_idemp
#[cfg(test)]
static CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
pub(crate) async fn delete_confirmed_transition_candidate_exact_with_manager_and_identity(
obj_name: &str,
rv_id: &str,
crates/ecstore/src/bucket/lifecycle/transition_transaction.rs
+412 -64
View File
@@ -25,14 +25,17 @@ use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPAC
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::lifecycle::tier_sweeper::{
delete_confirmed_transition_candidate_exact_with_lease_idempotent,
delete_confirmed_transition_candidate_exact_with_manager_and_identity,
delete_object_from_remote_tier_idempotent_with_manager_and_identity,
};
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::{Error, Result as EcstoreResult};
use crate::object_api::ObjectOptions;
use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe};
use crate::storage_api_contracts::{list::ListOperations as _, object::ObjectOperations as _};
use crate::storage_api_contracts::{
list::ListOperations as _,
namespace::NamespaceLocking as _,
object::{HTTPPreconditions, ObjectOperations as _},
};
use crate::store::ECStore;
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -110,7 +113,7 @@ pub struct TransitionRemoteVersion {
impl TransitionRemoteVersion {
pub fn known_from_put_response(version_id: impl Into<String>) -> Self {
let version_id = version_id.into();
if version_id.is_empty() || Uuid::parse_str(&version_id).is_ok_and(|parsed| parsed.is_nil()) {
if version_id.is_empty() {
Self::unversioned()
} else {
Self::versioned(version_id)
@@ -307,10 +310,16 @@ impl TransitionTransaction {
if self.write_id.is_nil() {
return Err(TransitionTransactionError::Corrupt("write_id is nil"));
}
if self.not_after_unix_nanos <= 0 {
return Err(TransitionTransactionError::Corrupt("ownership deadline is not positive"));
}
self.source.validate()?;
if self.tier_name.is_empty() {
return Err(TransitionTransactionError::Corrupt("tier name is empty"));
}
if self.backend_fingerprint == [0; 32] {
return Err(TransitionTransactionError::Corrupt("backend fingerprint is empty"));
}
if self.remote_object
!= canonical_transition_remote_object(self.deployment_id, &self.source.bucket, self.transaction_id, self.write_id)?
{
@@ -476,6 +485,18 @@ impl TransitionTransaction {
Ok(())
}
fn has_same_immutable_identity(&self, other: &Self) -> bool {
self.deployment_id == other.deployment_id
&& self.transaction_id == other.transaction_id
&& self.owner_epoch == other.owner_epoch
&& self.write_id == other.write_id
&& self.source == other.source
&& self.tier_name == other.tier_name
&& self.backend_fingerprint == other.backend_fingerprint
&& self.remote_object == other.remote_object
&& self.not_after_unix_nanos == other.not_after_unix_nanos
}
fn validate_cleanup_proof(&self, proof: &TransitionCleanupProof) -> Result<()> {
if proof.transaction_id != self.transaction_id
|| proof.write_id != self.write_id
@@ -585,20 +606,91 @@ pub(crate) async fn save_transition_transaction_record(
let object =
transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
let data = transaction.encode().map_err(transition_transaction_store_error)?;
config_boundary::save_config(api.clone(), &object, data.clone()).await?;
config_boundary::save_config_with_opts(
api.clone(),
&object,
data.clone(),
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
if_none_match: Some("*".to_string()),
..Default::default()
}),
..Default::default()
},
)
.await?;
// Box::pin: the durable-receipt state machine is large and sits on the
// already-deep transition worker poll chain; keeping it inline overflows
// the default 2 MiB tokio worker stack in debug builds.
Box::pin(api.record_durable_ilm_decommission_progress(&object, &data)).await
}
pub(crate) async fn save_transition_transaction_record_if_current(
api: Arc<ECStore>,
expected: &TransitionTransaction,
next: &TransitionTransaction,
) -> EcstoreResult<()> {
let object = transition_transaction_record_object_name(next.transaction_id).map_err(transition_transaction_store_error)?;
let revision_is_next = expected.revision.checked_add(1) == Some(next.revision);
let state_is_next = state_change_allowed(expected.state, next.state)
|| matches!(
(expected.state, next.state),
(
TransitionTransactionState::Uploaded
| TransitionTransactionState::UploadOutcomeUnknown
| TransitionTransactionState::LocalCommitStarted,
TransitionTransactionState::CleanupPending
)
);
let remote_version_is_monotonic = expected.remote_version.is_unknown() || expected.remote_version == next.remote_version;
if !expected.has_same_immutable_identity(next) || !revision_is_next || !state_is_next || !remote_version_is_monotonic {
return Err(Error::PreconditionFailed);
}
let (current, etag) = load_transition_transaction_record_with_etag(api.clone(), expected.transaction_id).await?;
if &current != expected {
return Err(Error::PreconditionFailed);
}
let data = next.encode().map_err(transition_transaction_store_error)?;
config_boundary::save_config_with_opts(
api.clone(),
&object,
data.clone(),
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
if_match: Some(etag),
..Default::default()
}),
..Default::default()
},
)
.await?;
// Box::pin: see save_transition_transaction_record.
Box::pin(api.record_durable_ilm_decommission_progress(&object, &data)).await
}
pub(crate) async fn load_transition_transaction_record(
api: Arc<ECStore>,
transaction_id: Uuid,
) -> EcstoreResult<TransitionTransaction> {
load_transition_transaction_record_with_etag(api, transaction_id)
.await
.map(|(transaction, _)| transaction)
}
async fn load_transition_transaction_record_with_etag(
api: Arc<ECStore>,
transaction_id: Uuid,
) -> EcstoreResult<(TransitionTransaction, String)> {
let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?;
let data = config_boundary::read_config(api, &object).await?;
TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)
let (data, object_info) = config_boundary::read_config_with_metadata(api, &object, &ObjectOptions::default()).await?;
let etag = object_info
.etag
.filter(|etag| !etag.trim().is_empty())
.ok_or_else(|| Error::other("transition transaction record is missing an ETag"))?;
let transaction = TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)?;
Ok((transaction, etag))
}
pub(crate) async fn delete_transition_transaction_record(
@@ -607,10 +699,18 @@ pub(crate) async fn delete_transition_transaction_record(
) -> EcstoreResult<()> {
let object =
transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
let data = transaction.encode().map_err(transition_transaction_store_error)?;
let (current, etag) = match load_transition_transaction_record_with_etag(api.clone(), transaction.transaction_id).await {
Ok(record) => record,
Err(Error::ConfigNotFound) => return Ok(()),
Err(err) => return Err(err),
};
if &current != transaction {
return Err(Error::PreconditionFailed);
}
let data = current.encode().map_err(transition_transaction_store_error)?;
// Box::pin: see save_transition_transaction_record.
Box::pin(api.record_durable_ilm_decommission_terminal(&object, &data)).await?;
match config_boundary::delete_config(api, &object).await {
match config_boundary::delete_config_if_match(api, &object, &etag).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
Err(err) => Err(err),
}
@@ -637,6 +737,84 @@ pub enum TransitionTransactionRecoveryOutcome {
Retained,
}
#[cfg(test)]
#[derive(Default)]
struct TransitionRecoveryClaimBarrierState {
transaction_id: Uuid,
arrived: tokio::sync::Notify,
release: tokio::sync::Notify,
}
#[cfg(test)]
pub(crate) struct TransitionRecoveryClaimBarrier {
state: Arc<TransitionRecoveryClaimBarrierState>,
}
#[cfg(test)]
static TRANSITION_RECOVERY_CLAIM_BARRIER: std::sync::OnceLock<
std::sync::Mutex<Option<Arc<TransitionRecoveryClaimBarrierState>>>,
> = std::sync::OnceLock::new();
#[cfg(test)]
impl TransitionRecoveryClaimBarrier {
pub(crate) fn install(transaction_id: Uuid) -> Self {
let state = Arc::new(TransitionRecoveryClaimBarrierState {
transaction_id,
..Default::default()
});
let mut slot = TRANSITION_RECOVERY_CLAIM_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition recovery claim barrier mutex should not poison");
assert!(
slot.is_none(),
"transition recovery claim 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 reach the cleanup claim CAS");
}
pub(crate) fn release(&self) {
self.state.release.notify_one();
}
}
#[cfg(test)]
impl Drop for TransitionRecoveryClaimBarrier {
fn drop(&mut self) {
self.state.release.notify_one();
let mut slot = TRANSITION_RECOVERY_CLAIM_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition recovery claim 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_before_transition_recovery_claim(transaction_id: Uuid) {
let barrier = TRANSITION_RECOVERY_CLAIM_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition recovery claim 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 {
@@ -858,49 +1036,129 @@ pub async fn process_transition_transaction_record(
transaction: &TransitionTransaction,
) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
transaction.validate().map_err(transition_transaction_store_error)?;
match transaction.state {
// Box the expanded recovery state machine so callers on Tokio's default
// worker stack do not inline its full future into an already-deep scan.
Box::pin(process_transition_transaction_record_at(
api,
transaction,
time::OffsetDateTime::now_utc().unix_timestamp_nanos(),
))
.await
}
async fn process_transition_transaction_record_at(
api: Arc<ECStore>,
observed: &TransitionTransaction,
now_unix_nanos: i128,
) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
let record_name =
transition_transaction_record_object_name(observed.transaction_id).map_err(transition_transaction_store_error)?;
// 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.
let recovery_lock = api
.new_ns_lock(RUSTFS_META_BUCKET, &format!("{record_name}.recovery-lock"))
.await?;
let _recovery_guard = recovery_lock
.get_write_lock(crate::set_disk::get_lock_acquire_timeout())
.await?;
let current = match load_transition_transaction_record(api.clone(), observed.transaction_id).await {
Ok(current) => current,
Err(Error::ConfigNotFound) => return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted),
Err(err) => return Err(err),
};
if &current != observed {
return Ok(TransitionTransactionRecoveryOutcome::Retained);
}
match current.state {
TransitionTransactionState::Uploaded => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
if transition_transaction_ownership_is_active(&current, now_unix_nanos) {
return Ok(TransitionTransactionRecoveryOutcome::Retained);
}
Ok(false) => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
Err(err) if transition_source_is_missing(&err) => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
Err(err) => Err(err),
},
TransitionTransactionState::LocalCommitStarted => {
match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
Err(err) if transition_source_is_missing(&err) => 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(), &current, &cleanup).await {
Ok(()) => recover_cleanup_pending(api, &cleanup).await,
Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained),
Err(err) => Err(err),
}
}
TransitionTransactionState::CleanupPending => recover_cleanup_pending(api, &current).await,
TransitionTransactionState::LocalCommitStarted => match local_commit_matches_transaction(api.clone(), &current).await {
Ok(true) => {
delete_transition_transaction_record(api, &current).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::Retained),
Err(err) => Err(err),
},
TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
delete_transition_transaction_record(api, &current).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionTransactionState::UploadOutcomeUnknown => {
if transition_transaction_ownership_is_active(&current, now_unix_nanos) {
Ok(TransitionTransactionRecoveryOutcome::Retained)
} else {
recover_unknown_upload_outcome(api, &current).await
}
}
TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained),
}
}
fn transition_transaction_ownership_is_active(transaction: &TransitionTransaction, now_unix_nanos: i128) -> bool {
now_unix_nanos < i128::from(transaction.not_after_unix_nanos)
}
async fn recover_cleanup_pending(
api: Arc<ECStore>,
transaction: &TransitionTransaction,
) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await,
TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained),
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),
}
}
async fn delete_unreferenced_transition_candidate(
api: Arc<ECStore>,
transaction: &TransitionTransaction,
) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
let current = match load_transition_transaction_record(api.clone(), transaction.transaction_id).await {
Ok(current) => current,
Err(Error::ConfigNotFound) => return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted),
Err(err) => return Err(err),
};
if &current != transaction || current.state != TransitionTransactionState::CleanupPending {
return Ok(TransitionTransactionRecoveryOutcome::Retained);
}
delete_transition_remote_candidate(api.clone(), &current).await?;
delete_transition_transaction_record(api, &current).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
async fn recover_unknown_upload_outcome(
api: Arc<ECStore>,
transaction: &TransitionTransaction,
@@ -928,17 +1186,7 @@ async fn recover_unknown_upload_outcome(
TransitionCandidateProbe::VersionedPresent(version_id)
if Uuid::parse_str(&version_id).is_ok_and(|version_id| version_id.is_nil()) =>
{
delete_confirmed_transition_candidate_exact_with_manager_and_identity(
&transaction.remote_object,
&version_id,
&transaction.tier_name,
transaction.backend_fingerprint,
&api.tier_config_mgr(),
)
.await
.map_err(Error::other)?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
Ok(TransitionTransactionRecoveryOutcome::Retained)
}
TransitionCandidateProbe::VersionedPresent(version_id) => {
cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await
@@ -968,10 +1216,11 @@ async fn cleanup_recovered_unknown_upload_candidate(
},
)
.map_err(transition_transaction_store_error)?;
save_transition_transaction_record(api.clone(), &cleanup).await?;
delete_transition_remote_candidate(api.clone(), &cleanup).await?;
delete_transition_transaction_record(api, &cleanup).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
match save_transition_transaction_record_if_current(api.clone(), transaction, &cleanup).await {
Ok(()) => recover_cleanup_pending(api, &cleanup).await,
Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained),
Err(err) => Err(err),
}
}
fn transition_source_is_missing(err: &Error) -> bool {
@@ -986,13 +1235,7 @@ fn transition_source_is_missing(err: &Error) -> bool {
}
async fn local_commit_matches_transaction(api: Arc<ECStore>, transaction: &TransitionTransaction) -> EcstoreResult<bool> {
let opts = ObjectOptions {
version_id: transaction.source.version_id.map(|version_id| version_id.to_string()),
versioned: transaction.source.version_mode == TransitionSourceVersionMode::Versioned,
version_suspended: transaction.source.version_mode == TransitionSourceVersionMode::VersionSuspended,
metadata_cache_safe: false,
..Default::default()
};
let opts = transition_source_lookup_options(transaction);
let object = api
.get_object_info(&transaction.source.bucket, &transaction.source.object, &opts)
.await?;
@@ -1003,6 +1246,23 @@ async fn local_commit_matches_transaction(api: Arc<ECStore>, transaction: &Trans
&& transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_default())
}
fn transition_source_lookup_options(transaction: &TransitionTransaction) -> ObjectOptions {
ObjectOptions {
version_id: match transaction.source.version_mode {
TransitionSourceVersionMode::Versioned => transaction.source.version_id.map(|version_id| version_id.to_string()),
// Both modes identify the stored null version. Query it explicitly
// so a later versioning change cannot redirect the proof to a new latest version.
TransitionSourceVersionMode::Unversioned | TransitionSourceVersionMode::VersionSuspended => {
Some(Uuid::nil().to_string())
}
},
versioned: transaction.source.version_mode == TransitionSourceVersionMode::Versioned,
version_suspended: transaction.source.version_mode == TransitionSourceVersionMode::VersionSuspended,
metadata_cache_safe: false,
..Default::default()
}
}
async fn delete_transition_remote_candidate(api: Arc<ECStore>, transaction: &TransitionTransaction) -> EcstoreResult<()> {
let version_id = transaction.remote_version.tier_delete_version_id().unwrap_or_default();
let version_id_exact = transaction.remote_version.kind == TransitionRemoteVersionKind::Versioned;
@@ -1023,6 +1283,25 @@ pub async fn recover_transition_transaction_records(
api: Arc<ECStore>,
limit: usize,
marker: Option<String>,
) -> EcstoreResult<TransitionTransactionRecoveryStats> {
recover_transition_transaction_records_with_now(api, limit, marker, None).await
}
#[cfg(any(test, feature = "test-util"))]
pub async fn recover_transition_transaction_records_at(
api: Arc<ECStore>,
limit: usize,
marker: Option<String>,
now_unix_nanos: i128,
) -> EcstoreResult<TransitionTransactionRecoveryStats> {
recover_transition_transaction_records_with_now(api, limit, marker, Some(now_unix_nanos)).await
}
async fn recover_transition_transaction_records_with_now(
api: Arc<ECStore>,
limit: usize,
marker: Option<String>,
now_unix_nanos: Option<i128>,
) -> EcstoreResult<TransitionTransactionRecoveryStats> {
if limit == 0 {
return Err(Error::other("transition transaction recovery limit must be greater than zero"));
@@ -1087,7 +1366,13 @@ pub async fn recover_transition_transaction_records(
}
};
match process_transition_transaction_record(api.clone(), &transaction).await {
let recovery = match now_unix_nanos {
Some(now_unix_nanos) => {
Box::pin(process_transition_transaction_record_at(api.clone(), &transaction, now_unix_nanos)).await
}
None => process_transition_transaction_record(api.clone(), &transaction).await,
};
match recovery {
Ok(
TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted
| TransitionTransactionRecoveryOutcome::RecordDeleted,
@@ -1332,6 +1617,34 @@ mod tests {
.expect("expired unknown upload outcome should be eligible");
}
#[test]
fn transition_ownership_window_expires_at_not_after() {
let transaction = new_transaction();
let deadline = i128::from(transaction.not_after_unix_nanos);
assert!(transition_transaction_ownership_is_active(&transaction, deadline - 1));
assert!(!transition_transaction_ownership_is_active(&transaction, deadline));
}
#[test]
fn null_transition_source_lookup_targets_the_exact_version_shape() {
for mode in [
TransitionSourceVersionMode::Unversioned,
TransitionSourceVersionMode::VersionSuspended,
] {
let mut transaction = new_transaction();
transaction.source.version_id = None;
transaction.source.version_mode = mode;
let opts = transition_source_lookup_options(&transaction);
assert_eq!(opts.version_id, Some(Uuid::nil().to_string()));
assert!(!opts.versioned);
assert_eq!(opts.version_suspended, mode == TransitionSourceVersionMode::VersionSuspended);
assert!(!opts.metadata_cache_safe);
}
}
fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
TransitionCleanupProof {
transaction_id: transaction.transaction_id,
@@ -1346,10 +1659,17 @@ mod tests {
#[test]
fn remote_version_distinguishes_unknown_unversioned_and_versioned() {
assert_eq!(TransitionRemoteVersion::known_from_put_response("").tier_delete_version_id(), None);
let nil_version = Uuid::nil().to_string();
let invalid_nil = TransitionRemoteVersion::known_from_put_response(nil_version.clone());
assert_eq!(
TransitionRemoteVersion::known_from_put_response(Uuid::nil().to_string()).tier_delete_version_id(),
None
invalid_nil.tier_delete_version_id(),
Some(nil_version.as_str()),
"a non-empty version must never be downgraded to an unversioned DELETE"
);
assert!(matches!(
invalid_nil.validate(),
Err(TransitionTransactionError::Corrupt("versioned remote version is nil uuid"))
));
let version_id = Uuid::new_v4().to_string();
assert_eq!(
@@ -1468,6 +1788,34 @@ mod tests {
})
.is_ok()
);
assert!(matches!(
TransitionTransaction::new(TransitionTransactionInit {
deployment_id: Uuid::new_v4(),
transaction_id: Uuid::new_v4(),
owner_epoch: Uuid::new_v4(),
write_id: Uuid::new_v4(),
source: source_identity(TransitionSourceVersionMode::Unversioned),
tier_name: "warm-tier".to_string(),
backend_fingerprint: [0; 32],
not_after_unix_nanos: 1,
}),
Err(TransitionTransactionError::Corrupt("backend fingerprint is empty"))
));
assert!(matches!(
TransitionTransaction::new(TransitionTransactionInit {
deployment_id: Uuid::new_v4(),
transaction_id: Uuid::new_v4(),
owner_epoch: Uuid::new_v4(),
write_id: Uuid::new_v4(),
source: source_identity(TransitionSourceVersionMode::Unversioned),
tier_name: "warm-tier".to_string(),
backend_fingerprint: BACKEND_FINGERPRINT,
not_after_unix_nanos: 0,
}),
Err(TransitionTransactionError::Corrupt("ownership deadline is not positive"))
));
}
#[test]
crates/ecstore/src/set_disk/mod.rs
+2
View File
@@ -876,6 +876,8 @@ pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
pub(crate) use ops::object::DeleteObjectCommitBarrier;
#[cfg(any(test, feature = "test-util"))]
pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
#[cfg(test)]
pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransitionUploadedCommitBarrier;
pub(crate) use ops::object::body_cache_plaintext_len;
#[cfg(all(test, feature = "test-util"))]
pub(crate) use ops::object::cleanup_rejected_transition_upload_durably;
crates/ecstore/src/set_disk/ops/object.rs
+212 -64
View File
@@ -266,9 +266,9 @@ use crate::bucket::lifecycle::{
transitioned_force_delete_journal_entry,
},
transition_transaction::{
TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
load_transition_transaction_record, save_transition_transaction_record,
TransitionCleanupDecision, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode,
TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
save_transition_transaction_record, save_transition_transaction_record_if_current,
},
};
use crate::bucket::quota::reservation;
@@ -5513,6 +5513,7 @@ pub(crate) struct TransitionUploadCleanup {
object: String,
candidate: Option<TransitionUploadCandidate>,
cleanup_api: Option<Arc<ECStore>>,
cleanup_transaction: Option<TransitionTransaction>,
armed: bool,
}
@@ -5523,10 +5524,20 @@ impl TransitionUploadCleanup {
object: object.to_string(),
candidate: None,
cleanup_api: None,
cleanup_transaction: None,
armed: true,
}
}
fn set_cleanup_owner(&mut self, api: Option<Arc<ECStore>>, transaction: &TransitionTransaction) {
self.cleanup_api = api;
self.cleanup_transaction = Some(transaction.clone());
}
fn update_cleanup_transaction(&mut self, transaction: &TransitionTransaction) {
self.cleanup_transaction = Some(transaction.clone());
}
fn cleanup_candidate(&self) -> std::io::Result<&TransitionUploadCandidate> {
self.candidate
.as_ref()
@@ -5559,11 +5570,13 @@ impl TransitionUploadCleanup {
api: Option<Arc<ECStore>>,
transaction: &mut TransitionTransaction,
) -> std::io::Result<()> {
let api = api.or_else(|| self.cleanup_api.clone());
self.cleanup_api = api.clone();
let candidate = self.cleanup_candidate()?.clone();
let owner_error = persist_rejected_transition_cleanup_owner(api.as_ref(), transaction, &candidate)
.await
.err();
self.update_cleanup_transaction(transaction);
let cleanup = cleanup_rejected_transition_upload_durably(
&self.lease,
&self.object,
@@ -5615,15 +5628,37 @@ impl Drop for TransitionUploadCleanup {
}
};
let object = self.object.clone();
let cleanup_version = candidate.cleanup_version().to_string();
let version_id_exact = candidate.cleanup_version_is_exact();
let candidate = candidate.clone();
let cleanup_api = self.cleanup_api.clone();
let mut cleanup_transaction = self.cleanup_transaction.clone();
if let Ok(handle) = tokio::runtime::Handle::try_current() {
handle.spawn(async move {
if let Err(err) =
cleanup_rejected_transition_upload_durably(&lease, &object, &cleanup_version, version_id_exact, cleanup_api)
.await
{
let owner_error = match (cleanup_api.as_ref(), cleanup_transaction.as_mut()) {
(Some(api), Some(transaction)) => {
persist_rejected_transition_cleanup_owner(Some(api), transaction, &candidate)
.await
.err()
}
_ => None,
};
let cleanup_version = candidate.cleanup_version().to_string();
let cleanup = cleanup_rejected_transition_upload_durably(
&lease,
&object,
&cleanup_version,
candidate.cleanup_version_is_exact(),
cleanup_api,
)
.await;
let result = match (owner_error, cleanup) {
(_, Ok(())) => Ok(()),
(None, Err(cleanup_error)) => Err(cleanup_error),
(Some(owner_error), Err(cleanup_error)) => Err(crate::error::stable_io_error(
"cancelled transition upload cleanup error followed a transaction owner update failure",
format!("owner error: {owner_error}; cleanup error: {cleanup_error}"),
)),
};
if let Err(err) = result {
warn!(
event = EVENT_LIFECYCLE_TRANSITION_CLEANUP,
component = LOG_COMPONENT_ECSTORE,
@@ -5646,25 +5681,36 @@ async fn persist_rejected_transition_cleanup_owner(
transaction: &mut TransitionTransaction,
candidate: &TransitionUploadCandidate,
) -> Result<()> {
match transaction.state {
TransitionTransactionState::UploadOutcomeUnknown => {
transaction
.advance(
transaction.fence(),
TransitionTransactionState::Uploaded,
Some(TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string())),
)
.map_err(Error::other)?;
}
TransitionTransactionState::Uploaded => {}
let expected = transaction.clone();
let remote_version = TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string());
let decision = match transaction.state {
TransitionTransactionState::UploadOutcomeUnknown => TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation,
TransitionTransactionState::Uploaded => TransitionCleanupDecision::UploadAbortedBeforeLocalCommit,
TransitionTransactionState::CleanupPending if transaction.remote_version == remote_version => return Ok(()),
state => {
return Err(Error::other_with_context(
"transition transaction state cannot own a rejected upload",
format!("state {state:?}"),
));
}
}
save_transition_transaction_if_available(api, transaction).await
};
let mut cleanup = expected.clone();
cleanup
.mark_cleanup_pending(
expected.fence(),
crate::bucket::lifecycle::transition_transaction::TransitionCleanupProof {
transaction_id: expected.transaction_id,
write_id: expected.write_id,
remote_object: expected.remote_object.clone(),
remote_version,
backend_fingerprint: expected.backend_fingerprint,
decision,
},
)
.map_err(Error::other)?;
compare_and_save_transition_transaction_if_available(api, &expected, &cleanup).await?;
*transaction = cleanup;
Ok(())
}
pub(crate) async fn cleanup_rejected_transition_upload_durably(
@@ -5779,6 +5825,26 @@ async fn save_transition_transaction_if_available(api: Option<&Arc<ECStore>>, tr
}
}
async fn compare_and_save_transition_transaction_if_available(
api: Option<&Arc<ECStore>>,
expected: &TransitionTransaction,
next: &TransitionTransaction,
) -> Result<()> {
if let Some(api) = api {
// The transition worker already has a deep poll chain. Keep the CAS
// read/write/receipt future off Tokio's default worker stack.
return Box::pin(save_transition_transaction_record_if_current(api.clone(), expected, next)).await;
}
#[cfg(test)]
{
Ok(())
}
#[cfg(not(test))]
{
Err(Error::other("transition transaction store is unavailable"))
}
}
async fn advance_and_save_transition_transaction(
api: Option<&Arc<ECStore>>,
transaction: &mut TransitionTransaction,
@@ -5787,10 +5853,14 @@ async fn advance_and_save_transition_transaction(
) -> Result<()> {
#[cfg(test)]
record_transition_uploaded_save_attempt(transaction, next);
transaction
.advance(transaction.fence(), next, remote_version)
let expected = transaction.clone();
let mut advanced = expected.clone();
advanced
.advance(expected.fence(), next, remote_version)
.map_err(Error::other)?;
save_transition_transaction_if_available(api, transaction).await
compare_and_save_transition_transaction_if_available(api, &expected, &advanced).await?;
*transaction = advanced;
Ok(())
}
#[cfg(test)]
@@ -5874,29 +5944,29 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction,
}
}
async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
async fn delete_transition_transaction_if_available(
api: Option<&Arc<ECStore>>,
transaction: &TransitionTransaction,
) -> Result<()> {
if let Some(api) = api {
let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
Ok(transaction) => transaction,
Err(Error::ConfigNotFound) => return Ok(()),
Err(err) => return Err(err),
};
return delete_transition_transaction_record(api.clone(), &transaction).await;
// Conditional delete now includes a read and terminal receipt; box it
// for the same transition-worker stack bound as the CAS path above.
return Box::pin(delete_transition_transaction_record(api.clone(), transaction)).await;
}
Ok(())
}
async fn delete_transition_transaction_after_remote_cleanup(
api: Option<&Arc<ECStore>>,
transaction_id: Uuid,
transaction: &TransitionTransaction,
bucket: &str,
object: &str,
) {
if let Err(err) = delete_transition_transaction_if_available(api, transaction_id).await {
if let Err(err) = delete_transition_transaction_if_available(api, transaction).await {
warn!(
bucket = bucket,
object = object,
transaction_id = %transaction_id,
transaction_id = %transaction.transaction_id,
error = ?err,
"transition remote candidate was cleaned but transaction record cleanup failed"
);
@@ -6031,6 +6101,86 @@ async fn pause_after_transition_upload_candidate_recorded() {
}
}
#[cfg(test)]
struct TransitionUploadedCommitBarrierState {
bucket: String,
object: String,
arrived: tokio::sync::Notify,
release: tokio::sync::Notify,
}
#[cfg(test)]
pub(crate) struct TransitionUploadedCommitBarrier {
state: Arc<TransitionUploadedCommitBarrierState>,
}
#[cfg(test)]
static TRANSITION_UPLOADED_COMMIT_BARRIER: std::sync::OnceLock<
std::sync::Mutex<Option<Arc<TransitionUploadedCommitBarrierState>>>,
> = std::sync::OnceLock::new();
#[cfg(test)]
impl TransitionUploadedCommitBarrier {
pub(crate) fn install(bucket: &str, object: &str) -> Self {
let state = Arc::new(TransitionUploadedCommitBarrierState {
bucket: bucket.to_string(),
object: object.to_string(),
arrived: tokio::sync::Notify::new(),
release: tokio::sync::Notify::new(),
});
let mut slot = TRANSITION_UPLOADED_COMMIT_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition uploaded commit barrier mutex should not poison");
assert!(
slot.is_none(),
"transition uploaded commit 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 should persist Uploaded before acquiring its commit lock");
}
pub(crate) fn release(&self) {
self.state.release.notify_one();
}
}
#[cfg(test)]
impl Drop for TransitionUploadedCommitBarrier {
fn drop(&mut self) {
self.state.release.notify_one();
let mut slot = TRANSITION_UPLOADED_COMMIT_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition uploaded commit 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_uploaded_persisted(bucket: &str, object: &str) {
let barrier = TRANSITION_UPLOADED_COMMIT_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("transition uploaded commit barrier mutex should not poison")
.as_ref()
.filter(|barrier| barrier.bucket == bucket && barrier.object == object)
.cloned();
if let Some(barrier) = barrier {
barrier.arrived.notify_one();
barrier.release.notified().await;
}
}
#[cfg(test)]
#[derive(Clone, Copy, PartialEq, Eq)]
enum TransitionCommitPause {
@@ -8841,6 +8991,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
};
let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj);
upload_cleanup.set_cleanup_owner(transaction_api.clone(), &transaction);
advance_and_save_transition_transaction(
transaction_api.as_ref(),
&mut transaction,
@@ -8848,6 +8999,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
None,
)
.await?;
upload_cleanup.update_cleanup_transaction(&transaction);
let remote_upload = {
let lease = &upload_cleanup.lease;
let recorded_candidate = &mut upload_cleanup.candidate;
@@ -8872,7 +9024,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
failure.error
))));
}
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object)
.await;
}
return Err(failure.error);
@@ -8886,7 +9038,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
"{err}; rejected remote upload cleanup failed: {cleanup_err}"
))));
}
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
return Err(err.into());
}
let fleet_proof = remote_version_state_writer_fleet_proof();
@@ -8902,7 +9054,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
"{err}; rejected remote upload cleanup failed: {cleanup_err}"
))));
}
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object)
.await;
return Err(err.into());
}
@@ -8921,9 +9073,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
"{err}; uploaded transition transaction persist failed and cleanup failed: {cleanup_err}"
))));
}
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
return Err(err);
}
upload_cleanup.update_cleanup_transaction(&transaction);
#[cfg(test)]
pause_after_transition_uploaded_persisted(bucket, object).await;
let commit_opts = opts.as_commit_opts();
// Note: Using clone() here is necessary because ObjectOptions has 124 fields.
@@ -8937,7 +9093,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(
transaction_api.as_ref(),
transaction_id,
&transaction,
bucket,
object,
)
@@ -8954,7 +9110,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
Err(err) => {
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object)
.await;
}
return Err(err);
@@ -8970,8 +9126,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let already_transitioned = current_fi.transition_status == TRANSITION_COMPLETE;
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
}
if already_transitioned {
return Ok(());
@@ -9000,8 +9155,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if transition_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
}
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "transition_object_commit",
@@ -9016,8 +9170,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if !upload_cleanup.lease.is_current_generation() {
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
}
return Err(Error::other("remote tier configuration changed during transition"));
}
@@ -9033,8 +9186,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
{
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
}
return Err(Error::other("remote version state fleet capability changed during transition"));
}
@@ -9048,8 +9200,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
{
drop(transition_lock_guard);
if upload_cleanup.cleanup().await.is_ok() {
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
.await;
delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await;
}
return Err(err);
}
@@ -9065,19 +9216,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
drop(transition_lock_guard);
return Err(err);
}
match transaction.advance(transaction.fence(), TransitionTransactionState::Committed, None) {
Ok(_) => {
if let Err(err) = save_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await {
warn!(
bucket = bucket,
object = object,
transaction_id = %transaction_id,
error = ?err,
"transition committed locally but transaction committed-state persist failed"
);
} else if let Err(err) =
delete_transition_transaction_if_available(transaction_api.as_ref(), transaction_id).await
{
match advance_and_save_transition_transaction(
transaction_api.as_ref(),
&mut transaction,
TransitionTransactionState::Committed,
None,
)
.await
{
Ok(()) => {
if let Err(err) = delete_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await {
warn!(
bucket = bucket,
object = object,
crates/ecstore/src/store/init.rs
+319 -34
View File
@@ -832,11 +832,12 @@ mod tests {
},
transition_transaction::{
TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError,
TransitionOperatorProbe, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode,
TransitionTransaction, TransitionTransactionInit, TransitionTransactionState,
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, save_transition_transaction_record,
recover_transition_transaction_records, recover_transition_transaction_records_at,
save_transition_transaction_record, save_transition_transaction_record_if_current,
transition_transaction_record_object_name,
},
validate_durable_ilm_record,
@@ -864,6 +865,7 @@ mod tests {
tier_mutation_peer::{TierMutationPeerError, TierMutationPeerState, handle_tier_mutation_peer_request},
warm_backend::{TransitionCandidateProbe, WarmBackend},
},
set_disk::SetDiskTransitionUploadedCommitBarrier as TransitionUploadedCommitBarrier,
storage_api_contracts::list::ListOperations as _,
};
#[cfg(feature = "test-util")]
@@ -17469,7 +17471,82 @@ mod tests {
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn transition_transaction_recovery_deletes_uploaded_remote_candidate() {
async fn transition_transaction_recovery_retains_active_uploaded_candidate_until_commit() {
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-active-uploaded", &[4]))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let tier_name = "TXACTIVEUPLOADED";
let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
let bucket = "transition-transaction-active-uploaded-bucket";
let object = "object.bin";
let payload = b"active Uploaded ownership must survive recovery until local commit".repeat(1024);
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("source bucket should be created");
let mut reader = PutObjReader::from_vec(payload.clone());
let source = store
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("source object should be written");
let barrier = TransitionUploadedCommitBarrier::install(bucket, object);
let transition_store = store.clone();
let transition = tokio::spawn(async move {
transition_store
.transition_object(
bucket,
object,
&ObjectOptions {
transition: TransitionOptions {
status: TRANSITION_PENDING.to_string(),
tier: tier_name.to_string(),
etag: source.etag.clone().expect("source object should have an ETag"),
..Default::default()
},
version_id: source.version_id.map(|version_id| version_id.to_string()),
mod_time: source.mod_time,
..Default::default()
},
)
.await
});
barrier.wait_until_paused().await;
let stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("recovery should inspect the active Uploaded transaction");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0));
assert_eq!(backend.object_count().await, 1, "active ownership must retain the remote candidate");
assert_eq!(backend.remove_count().await, 0, "active ownership must fence remote DELETE");
assert_eq!(transition_transaction_record_count(store.clone()).await, 1);
barrier.release();
transition
.await
.expect("transition task should join")
.expect("transition should commit after the barrier is released");
assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
assert_eq!(backend.remove_count().await, 0);
let mut body = Vec::new();
store
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("committed tier object should remain readable")
.stream
.read_to_end(&mut body)
.await
.expect("tier object body should drain");
assert_eq!(body, payload);
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn transition_transaction_recovery_deletes_expired_uploaded_remote_candidate() {
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-recovery", &[4])).await;
@@ -17477,12 +17554,114 @@ mod tests {
let tier_name = "TXRECOVERY";
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 exact_version = uuid::Uuid::new_v4().to_string();
let cases = [
(
"exact",
exact_version.clone(),
TransitionRemoteVersion::versioned(exact_version),
TransitionSourceVersionMode::Versioned,
),
(
"suspended-null",
"null".to_string(),
TransitionRemoteVersion::versioned("null"),
TransitionSourceVersionMode::VersionSuspended,
),
(
"known-unversioned",
String::new(),
TransitionRemoteVersion::unversioned(),
TransitionSourceVersionMode::Unversioned,
),
];
let mut expected_removes = 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"),
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: format!("source-{case}"),
version_id: (source_mode == TransitionSourceVersionMode::Versioned).then(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: source_mode,
},
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(remote_version))
.expect("transaction should enter uploaded state");
backend.set_put_remote_version(Some(put_version.clone())).await;
let candidate = bytes::Bytes::from_static(b"orphan 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");
expected_removes.push((transaction.remote_object, put_version));
}
let stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("transition transaction recovery should run");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (3, 3, 0, 0));
assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
let mut actual_removes = backend.remove_versions().await;
actual_removes.sort();
expected_removes.sort();
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);
let replay = recover_transition_transaction_records(store, 100, None)
.await
.expect("replayed recovery should remain idempotent");
assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (0, 0, 0, 0));
assert_eq!(
backend.remove_versions().await.len(),
3,
"each expired candidate must be deleted only once"
);
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn transition_transaction_recovery_cas_loses_to_active_state_advance() {
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-recovery-cas", &[4]))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let tier_name = "TXRECOVERYCAS";
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 {
let mut uploaded = 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(),
@@ -17499,7 +17678,97 @@ mod tests {
},
tier_name: tier_name.to_string(),
backend_fingerprint: backend_identity,
not_after_unix_nanos: 1_780_000_000_000_000_000,
not_after_unix_nanos: 1,
})
.expect("transaction should build");
uploaded
.advance(
uploaded.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"CAS-owned transition remote candidate");
backend
.put(
&uploaded.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(), &uploaded)
.await
.expect("transaction record should persist");
let barrier = TransitionRecoveryClaimBarrier::install(uploaded.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 mut active = uploaded.clone();
active
.advance(active.fence(), TransitionTransactionState::LocalCommitStarted, None)
.expect("active owner should advance to local commit");
save_transition_transaction_record_if_current(store.clone(), &uploaded, &active)
.await
.expect("active owner should win the persisted CAS");
barrier.release();
let stats = recovery
.await
.expect("recovery task should join")
.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)
.await
.expect("newer transaction revision must remain"),
active
);
assert_eq!(backend.object_count().await, 1, "a stale recovery must not delete the candidate");
assert_eq!(backend.remove_count().await, 0);
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn transition_transaction_recovery_quarantines_missing_required_fields() {
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-recovery-corrupt",
&[4],
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let tier_name = "TXRECOVERYCORRUPT";
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: None,
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::Unversioned,
},
tier_name: tier_name.to_string(),
backend_fingerprint: backend_identity,
not_after_unix_nanos: 1,
})
.expect("transaction should build");
transaction
@@ -17509,8 +17778,8 @@ mod tests {
Some(TransitionRemoteVersion::versioned(remote_version.clone())),
)
.expect("transaction should enter uploaded state");
backend.set_put_remote_version(Some(remote_version.clone())).await;
let candidate = bytes::Bytes::from_static(b"orphan candidate");
backend.set_put_remote_version(Some(remote_version)).await;
let candidate = bytes::Bytes::from_static(b"corrupt journal candidate");
backend
.put(
&transaction.remote_object,
@@ -17519,23 +17788,34 @@ mod tests {
)
.await
.expect("mock backend should accept candidate");
save_transition_transaction_record(store.clone(), &transaction)
.await
.expect("transaction record should persist");
let mut persisted: serde_json::Value = serde_json::from_slice(&transaction.encode().expect("transaction should encode"))
.expect("transaction should be JSON");
persisted["transaction"]
.as_object_mut()
.expect("transaction payload should be an object")
.remove("not_after_unix_nanos");
let path =
transition_transaction_record_object_name(transaction.transaction_id).expect("transaction path should be canonical");
com::save_config(
store.clone(),
&path,
serde_json::to_vec(&persisted).expect("corrupt fixture should encode"),
)
.await
.expect("corrupt transaction fixture should persist");
let stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("transition transaction recovery should run");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0));
assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
.expect("recovery scan should isolate a corrupt record");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 0, 1));
assert_eq!(
backend.remove_versions().await,
vec![(transaction.remote_object.clone(), remote_version)],
"recovery must delete the exact uploaded candidate"
transition_transaction_record_count(store).await,
1,
"corrupt evidence must remain quarantined"
);
assert_eq!(backend.exact_remove_count(), 1);
assert_eq!(backend.object_count().await, 0);
assert_eq!(backend.object_count().await, 1, "corrupt evidence must never authorize remote DELETE");
assert_eq!(backend.remove_count().await, 0);
}
#[cfg(feature = "test-util")]
@@ -18130,14 +18410,14 @@ mod tests {
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn transition_transaction_recovery_deletes_provider_recovered_unknown_upload() {
async fn transition_transaction_recovery_handles_provider_recovered_unknown_upload() {
let versioned_remote = uuid::Uuid::new_v4().to_string();
let nil_remote = uuid::Uuid::nil().to_string();
for (case, tier_name, remote_version) in [
("missing", "TXPROBEMISSING", None),
("unversioned", "TXPROBEUNVERSIONED", Some(String::new())),
("versioned", "TXPROBEVERSIONED", Some(versioned_remote)),
("nil-version", "TXPROBENILVERSION", Some(nil_remote)),
for (case, tier_name, remote_version, should_recover) in [
("missing", "TXPROBEMISSING", None, true),
("unversioned", "TXPROBEUNVERSIONED", Some(String::new()), true),
("versioned", "TXPROBEVERSIONED", Some(versioned_remote), true),
("nil-version", "TXPROBENILVERSION", Some(nil_remote), false),
] {
let temp_dir = tempfile::tempdir().expect("create temp store dir");
let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store(
@@ -18170,7 +18450,7 @@ mod tests {
},
tier_name: tier_name.to_string(),
backend_fingerprint: backend_identity,
not_after_unix_nanos: 1_780_000_000_000_000_000,
not_after_unix_nanos: 1,
})
.expect("transaction should build");
transaction
@@ -18197,14 +18477,18 @@ mod tests {
.await
.expect("transition transaction recovery should run");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0));
assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
assert_eq!(
(stats.scanned, stats.recovered, stats.retained, stats.failed),
if should_recover { (1, 1, 0, 0) } else { (1, 0, 1, 0) }
);
assert_eq!(transition_transaction_record_count(store.clone()).await, usize::from(!should_recover));
assert_eq!(
backend.object_count().await,
0,
"case {case}: recovered unknown upload candidate must be absent"
usize::from(!should_recover),
"case {case}: only a valid provider version state may be recovered destructively"
);
let removed = remote_version
.filter(|_| should_recover)
.map(|version| vec![(transaction.remote_object.clone(), version)])
.unwrap_or_default();
assert_eq!(
@@ -18478,9 +18762,10 @@ mod tests {
assert_eq!(backend.remove_count().await, 0, "unsupported recovery must not attempt cleanup");
backend.set_transition_candidate_probe_override(None).await;
let stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("provider-authoritative recovery should run");
let stats =
recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(transaction.not_after_unix_nanos) + 1)
.await
.expect("provider-authoritative recovery should run");
assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0));
assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
assert_eq!(backend.object_count().await, 0, "recovery must delete the provider-confirmed candidate");
crates/scanner/tests/lifecycle_integration_test.rs
+32 -5
View File
@@ -47,8 +47,8 @@ use storage_api::lifecycle::{
TransitionCleanupStoreBarrier, TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects,
expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry,
init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk,
path2_bucket_object_with_base_path, recover_transition_transaction_records, register_mock_tier_util, update_bucket_metadata,
wait_for_free_version_absence,
path2_bucket_object_with_base_path, recover_transition_transaction_records, recover_transition_transaction_records_at,
register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence,
};
static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
@@ -944,6 +944,14 @@ mod serial_tests {
.is_cancelled()
);
tokio::time::timeout(Duration::from_secs(5), async {
while backend.exact_remove_count() < 1 {
tokio::task::yield_now().await;
}
})
.await
.expect("Drop must persist cleanup ownership before attempting the exact remote delete");
let retained = tokio::time::timeout(Duration::from_secs(30), async {
loop {
let recovery = recover_transition_transaction_records(ecstore.clone(), 100, None)
@@ -1088,7 +1096,7 @@ mod serial_tests {
.expect_err("a versioned candidate must not commit to an unversioned tier");
match case {
CleanupCase::Persisted | CleanupCase::DeleteFallback => {
CleanupCase::Persisted => {
assert_eq!(backend.remove_versions().await, backend.put_versions().await);
assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate");
let recovery = recover_transition_transaction_records(ecstore.clone(), 100, None)
@@ -1102,6 +1110,25 @@ mod serial_tests {
.expect("successful reconciliation must remove every transition transaction");
assert_eq!(empty.scanned, 0);
}
CleanupCase::DeleteFallback => {
assert_eq!(backend.remove_versions().await, backend.put_versions().await);
assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate");
let retained = recover_transition_transaction_records(ecstore.clone(), 100, None)
.await
.expect("active unknown ownership must remain fenced after the transaction store was offline");
assert_eq!((retained.scanned, retained.recovered, retained.retained, retained.failed), (1, 0, 1, 0));
let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX)
.await
.expect("expired unknown ownership may use the provider's missing proof");
assert_eq!(
(recovered.scanned, recovered.recovered, recovered.retained, recovered.failed),
(1, 1, 0, 0)
);
let empty = recover_transition_transaction_records(ecstore.clone(), 100, None)
.await
.expect("expired missing-candidate reconciliation must remove the transaction");
assert_eq!(empty.scanned, 0);
}
CleanupCase::RetryPersisted => {
assert!(
!err.to_string().contains("journal retry error"),
@@ -1139,9 +1166,9 @@ mod serial_tests {
assert_eq!(retained.recovered, 0);
assert_eq!(retained.retained + retained.failed, 1);
backend.set_remove_failure(false);
let recovered = recover_transition_transaction_records(ecstore.clone(), 100, None)
let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX)
.await
.expect("recovery should delete the candidate after the backend becomes available");
.expect("expired recovery should delete the candidate after the backend becomes available");
assert_eq!(
(recovered.scanned, recovered.recovered, recovered.retained, recovered.failed),
(1, 1, 0, 0)
crates/scanner/tests/storage_api/mod.rs
+5 -2
View File
@@ -12,7 +12,9 @@
// See the License for the specific language governing permissions and
// limitations under the License.
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::transition_transaction::recover_transition_transaction_records;
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::transition_transaction::{
recover_transition_transaction_records, recover_transition_transaction_records_at,
};
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::{
bucket_lifecycle_audit::LcEventSrc,
bucket_lifecycle_ops::{
@@ -54,6 +56,7 @@ pub(crate) mod lifecycle {
assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object,
free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys,
init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path,
recover_transition_transaction_records, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence,
recover_transition_transaction_records, recover_transition_transaction_records_at, register_mock_tier_util,
update_bucket_metadata, wait_for_free_version_absence,
};
}