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

fix(ecstore): track ILM recovery across decommission

Browse Source
This commit is contained in:
overtrue
2026-08-22 02:23:58 +08:00
parent 619fae2512
commit ecd779b6f3
8 changed files with 973 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/durable_namespace.rs
+274 -8
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_utils::crypto::hex_sha256;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::{manual_transition_job, tier_delete_journal, transition_transaction};
@@ -89,6 +91,7 @@ pub(crate) struct ValidatedDurableIlmRecord {
pub(crate) namespace: &'static str,
pub(crate) id_kind: &'static str,
pub(crate) id: String,
pub(crate) checkpoint: DurableIlmRecordCheckpoint,
}
impl ValidatedDurableIlmRecord {
@@ -97,6 +100,180 @@ impl ValidatedDurableIlmRecord {
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub(crate) enum DurableIlmRecordCheckpoint {
TierDeleteJournal {
content_sha256: String,
identity_sha256: String,
committed: bool,
},
TransitionTransaction {
content_sha256: String,
identity_sha256: String,
remote_version_sha256: String,
remote_version_known: bool,
revision: u64,
state: transition_transaction::TransitionTransactionState,
},
ManualTransitionJob {
content_sha256: String,
identity_sha256: String,
updated_at_unix_nanos: i128,
state: manual_transition_job::ManualTransitionJobState,
scan_completed: bool,
cancel_requested: bool,
},
ManualTransitionScope {
content_sha256: String,
identity_sha256: String,
updated_at_unix_nanos: i128,
},
ManualTransitionTask {
content_sha256: String,
},
ManualTransitionWorkerResult {
content_sha256: String,
},
}
impl DurableIlmRecordCheckpoint {
pub(crate) fn content_sha256(&self) -> &str {
match self {
Self::TierDeleteJournal { content_sha256, .. }
| Self::TransitionTransaction { content_sha256, .. }
| Self::ManualTransitionJob { content_sha256, .. }
| Self::ManualTransitionScope { content_sha256, .. }
| Self::ManualTransitionTask { content_sha256 }
| Self::ManualTransitionWorkerResult { content_sha256 } => content_sha256,
}
}
pub(crate) fn validate_successor(&self, next: &Self) -> Result<()> {
if self == next {
return Ok(());
}
let valid = match (self, next) {
(
Self::TierDeleteJournal {
identity_sha256: previous_identity,
committed: previous_committed,
..
},
Self::TierDeleteJournal {
identity_sha256: next_identity,
committed: next_committed,
..
},
) => {
previous_identity == next_identity
&& (previous_committed == next_committed || (!previous_committed && *next_committed))
}
(
Self::TransitionTransaction {
identity_sha256: previous_identity,
remote_version_sha256: previous_remote_version,
remote_version_known: previous_remote_version_known,
revision: previous_revision,
state: previous_state,
..
},
Self::TransitionTransaction {
identity_sha256: next_identity,
remote_version_sha256: next_remote_version,
revision: next_revision,
state: next_state,
..
},
) => {
previous_identity == next_identity
&& transition_state_distance(*previous_state, *next_state)
.and_then(|distance| previous_revision.checked_add(distance))
.is_some_and(|expected_revision| *next_revision == expected_revision)
&& (!previous_remote_version_known || previous_remote_version == next_remote_version)
}
(
Self::ManualTransitionJob {
identity_sha256: previous_identity,
updated_at_unix_nanos: previous_updated_at,
state: previous_state,
scan_completed: previous_scan_completed,
cancel_requested: previous_cancel_requested,
..
},
Self::ManualTransitionJob {
identity_sha256: next_identity,
updated_at_unix_nanos: next_updated_at,
state: next_state,
scan_completed: next_scan_completed,
cancel_requested: next_cancel_requested,
..
},
) => {
previous_identity == next_identity
&& next_updated_at > previous_updated_at
&& manual_job_state_reaches(*previous_state, *next_state)
&& (!previous_scan_completed || *next_scan_completed)
&& (!previous_cancel_requested || *next_cancel_requested)
}
(
Self::ManualTransitionScope {
identity_sha256: previous_identity,
updated_at_unix_nanos: previous_updated_at,
..
},
Self::ManualTransitionScope {
identity_sha256: next_identity,
updated_at_unix_nanos: next_updated_at,
..
},
) => previous_identity == next_identity && next_updated_at > previous_updated_at,
_ => false,
};
if valid {
Ok(())
} else {
Err(Error::other("durable ILM record generation is not a monotonic successor"))
}
}
}
fn transition_state_distance(
from: transition_transaction::TransitionTransactionState,
to: transition_transaction::TransitionTransactionState,
) -> Option<u64> {
use transition_transaction::TransitionTransactionState::{
AbortedNoRemote, CleanupPending, Committed, LocalCommitStarted, UploadOutcomeUnknown, UploadStarted, Uploaded,
};
match (from, to) {
(UploadStarted, UploadOutcomeUnknown | AbortedNoRemote | Uploaded) => Some(1),
(UploadStarted, LocalCommitStarted | CleanupPending) => Some(2),
(UploadStarted, Committed) => Some(3),
(UploadOutcomeUnknown, Uploaded | CleanupPending) => Some(1),
(UploadOutcomeUnknown, LocalCommitStarted) => Some(2),
(UploadOutcomeUnknown, Committed) => Some(3),
(Uploaded, LocalCommitStarted | CleanupPending) => Some(1),
(Uploaded, Committed) => Some(2),
(LocalCommitStarted, Committed | CleanupPending) => Some(1),
_ => None,
}
}
fn manual_job_state_reaches(
from: manual_transition_job::ManualTransitionJobState,
to: manual_transition_job::ManualTransitionJobState,
) -> bool {
from == to || from == manual_transition_job::ManualTransitionJobState::Running
}
fn checkpoint_hash<T: Serialize>(value: &T) -> Result<String> {
let encoded = serde_json::to_vec(value).map_err(Error::other)?;
Ok(hex_sha256(&encoded, ToOwned::to_owned))
}
fn path_is_in_namespace(path: &str, namespace: &DurableIlmNamespace) -> bool {
let Some(suffix) = path.strip_prefix(namespace.prefix) else {
return false;
@@ -163,7 +340,8 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
)));
}
let (id_kind, id) = match namespace.kind {
let content_sha256 = hex_sha256(data, ToOwned::to_owned);
let (id_kind, id, checkpoint) = match namespace.kind {
DurableIlmRecordKind::TierDeleteJournal => {
let entry = tier_delete_journal::decode_tier_delete_journal_entry(data)?;
if tier_delete_journal::tier_delete_journal_object_name(&entry) != path {
@@ -173,12 +351,52 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
.strip_prefix(namespace.prefix)
.and_then(|suffix| suffix.strip_suffix(".json"))
.ok_or_else(|| Error::other("tier delete journal path is invalid"))?;
("operation_id", operation_id.to_string())
let identity_sha256 = checkpoint_hash(&(
&entry.obj_name,
&entry.version_id,
&entry.tier_name,
entry.backend_identity,
entry.version_id_exact,
entry.version_state,
&entry.source,
))?;
(
"operation_id",
operation_id.to_string(),
DurableIlmRecordCheckpoint::TierDeleteJournal {
content_sha256,
identity_sha256,
committed: entry.state == super::tier_sweeper::TierDeleteJournalState::Committed,
},
)
}
DurableIlmRecordKind::TransitionTransaction => {
let transaction = transition_transaction::decode_transition_transaction_record(path, data)
.map_err(|err| Error::other(err.to_string()))?;
("transaction_id", transaction.transaction_id.to_string())
let identity_sha256 = checkpoint_hash(&(
transaction.deployment_id,
transaction.transaction_id,
transaction.owner_epoch,
transaction.write_id,
&transaction.source,
&transaction.tier_name,
transaction.backend_fingerprint,
&transaction.remote_object,
transaction.not_after_unix_nanos,
))?;
let remote_version_sha256 = checkpoint_hash(&transaction.remote_version)?;
(
"transaction_id",
transaction.transaction_id.to_string(),
DurableIlmRecordCheckpoint::TransitionTransaction {
content_sha256,
identity_sha256,
remote_version_sha256,
remote_version_known: !transaction.remote_version.is_unknown(),
revision: transaction.revision,
state: transaction.state,
},
)
}
DurableIlmRecordKind::ManualTransitionJob => {
let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
@@ -188,9 +406,31 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
if canonical != path {
return Err(Error::other("manual transition job path is not canonical"));
}
manual_transition_job::ManualTransitionJobRecord::decode(job_id, data)
let job = manual_transition_job::ManualTransitionJobRecord::decode(job_id, data)
.map_err(|err| Error::other(err.to_string()))?;
("job_id", job_id.to_string())
let identity_sha256 = checkpoint_hash(&(
job.job_id,
&job.scope_key,
&job.bucket,
&job.prefix,
&job.tier,
job.dry_run,
job.max_objects,
job.max_duration,
job.created_at_unix_nanos,
))?;
(
"job_id",
job_id.to_string(),
DurableIlmRecordCheckpoint::ManualTransitionJob {
content_sha256,
identity_sha256,
updated_at_unix_nanos: job.updated_at_unix_nanos,
state: job.state,
scan_completed: job.scan_completed,
cancel_requested: job.cancel_requested,
},
)
}
DurableIlmRecordKind::ManualTransitionScope => {
let admission: manual_transition_job::ManualTransitionScopeAdmission =
@@ -201,7 +441,24 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
if canonical != path {
return Err(Error::other("manual transition scope content does not match its path"));
}
("job_id", admission.job_id.to_string())
let identity_sha256 = checkpoint_hash(&(
&admission.schema,
&admission.scope_key,
admission.job_id,
&admission.bucket,
&admission.prefix,
&admission.tier,
admission.dry_run,
))?;
(
"job_id",
admission.job_id.to_string(),
DurableIlmRecordCheckpoint::ManualTransitionScope {
content_sha256,
identity_sha256,
updated_at_unix_nanos: admission.updated_at_unix_nanos,
},
)
}
DurableIlmRecordKind::ManualTransitionTask => {
let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?;
@@ -212,7 +469,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
}
manual_transition_job::ManualTransitionTaskRecord::decode(job_id, &task_key, data)
.map_err(|err| Error::other(err.to_string()))?;
("job_id", job_id.to_string())
(
"job_id",
job_id.to_string(),
DurableIlmRecordCheckpoint::ManualTransitionTask { content_sha256 },
)
}
DurableIlmRecordKind::ManualTransitionWorkerResult => {
let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?;
@@ -223,7 +484,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
}
manual_transition_job::ManualTransitionWorkerResultRecord::decode(job_id, &task_key, data)
.map_err(|err| Error::other(err.to_string()))?;
("job_id", job_id.to_string())
(
"job_id",
job_id.to_string(),
DurableIlmRecordCheckpoint::ManualTransitionWorkerResult { content_sha256 },
)
}
};
@@ -231,6 +496,7 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
namespace: namespace.name,
id_kind,
id,
checkpoint,
})
}
crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs
+29 -19
View File
@@ -320,7 +320,7 @@ impl ManualTransitionJobRecord {
}
}
self.queue_snapshot = queue_snapshot;
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.advance_updated_at();
self.mark_terminal_if_worker_drained();
}
@@ -363,14 +363,14 @@ impl ManualTransitionJobRecord {
self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
self.report.tier_failure_by_reason = scan_tier_failure_by_reason;
self.queue_snapshot = queue_snapshot;
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.advance_updated_at();
self.mark_terminal_if_worker_drained();
true
}
pub fn mark_cancel_requested(&mut self) {
self.cancel_requested = true;
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.advance_updated_at();
}
pub fn claim_recovery_lease(&mut self, owner_id: impl Into<String>, queue_snapshot: ManualTransitionQueueSnapshot) {
@@ -384,7 +384,7 @@ impl ManualTransitionJobRecord {
pub fn abandon_recovery_lease(&mut self, lease_id: Uuid) {
if self.state == ManualTransitionJobState::Running && self.lease_id == lease_id {
self.lease_expires_at_unix_nanos = 0;
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.advance_updated_at();
}
}
@@ -402,7 +402,7 @@ impl ManualTransitionJobRecord {
pub fn renew_lease(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) {
let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.updated_at_unix_nanos = now;
self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
self.lease_expires_at_unix_nanos = manual_transition_job_lease_expires_at(now);
self.queue_snapshot = queue_snapshot;
}
@@ -467,9 +467,13 @@ impl ManualTransitionJobRecord {
}
fn mark_updated_terminal(&mut self) {
self.advance_updated_at();
self.completed_at_unix_nanos = Some(self.updated_at_unix_nanos);
}
fn advance_updated_at(&mut self) {
let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.updated_at_unix_nanos = now;
self.completed_at_unix_nanos = Some(now);
self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
}
fn mark_terminal_if_worker_drained(&mut self) {
@@ -1113,7 +1117,8 @@ pub fn manual_transition_scope_record_object_name(scope_key: &str) -> Result<Str
pub async fn save_manual_transition_job_record(api: Arc<ECStore>, job: &ManualTransitionJobRecord) -> EcstoreResult<()> {
let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
let data = job.encode().map_err(manual_transition_job_store_error)?;
config_boundary::save_config(api, &object, data).await
config_boundary::save_config(api.clone(), &object, data.clone()).await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
pub async fn load_manual_transition_job_record(api: Arc<ECStore>, job_id: Uuid) -> EcstoreResult<ManualTransitionJobRecord> {
@@ -1146,9 +1151,9 @@ pub async fn save_manual_transition_job_record_if_current(
let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
let data = job.encode().map_err(manual_transition_job_store_error)?;
config_boundary::save_config_with_opts_quiet(
api,
api.clone(),
&object,
data,
data.clone(),
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1158,7 +1163,8 @@ pub async fn save_manual_transition_job_record_if_current(
..Default::default()
},
)
.await
.await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
/// Applies a job-record mutation with optimistic concurrency control.
@@ -1596,9 +1602,9 @@ pub async fn save_manual_transition_scope_admission_if_absent(
let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(admission).map_err(Error::other)?;
config_boundary::save_config_with_opts(
api,
api.clone(),
&object,
data,
data.clone(),
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1608,7 +1614,8 @@ pub async fn save_manual_transition_scope_admission_if_absent(
..Default::default()
},
)
.await
.await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
pub async fn load_manual_transition_scope_admission(
@@ -1646,9 +1653,9 @@ pub async fn save_manual_transition_scope_admission_if_current(
let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(admission).map_err(Error::other)?;
match config_boundary::save_config_with_opts(
api,
api.clone(),
&object,
data,
data.clone(),
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1664,7 +1671,8 @@ pub async fn save_manual_transition_scope_admission_if_current(
Err(Error::PreconditionFailed)
}
result => result,
}
}?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
pub async fn claim_manual_transition_scope_admission(
@@ -1957,13 +1965,15 @@ pub async fn delete_manual_transition_scope_admission_if_current(
job_id: Uuid,
lease_id: Uuid,
) -> EcstoreResult<bool> {
let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag,
let (admission, etag) = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => (admission, etag),
Ok(_) => return Ok(false),
Err(Error::ConfigNotFound) => return Ok(true),
Err(err) => return Err(err),
};
let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(&admission).map_err(Error::other)?;
api.record_durable_ilm_decommission_terminal(&object, &data).await?;
match config_boundary::delete_config_if_match(api, &object, &etag).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(true),
Err(Error::PreconditionFailed) => Ok(false),
crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs
+9
View File
@@ -433,9 +433,18 @@ async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jen
)
.await?;
}
record_tier_delete_journal_decommission_terminal(&api, je).await?;
remove_tier_delete_journal_entry(api, je).await
}
async fn record_tier_delete_journal_decommission_terminal(api: &Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
let path = tier_delete_journal_object_name(je);
let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?;
api.record_durable_ilm_decommission_terminal(&path, &data)
.await
.map_err(std::io::Error::other)
}
fn object_info_references_tier_delete(info: &ObjectInfo, je: &Jentry) -> std::io::Result<bool> {
if info.transitioned_object.status != rustfs_filemeta::TRANSITION_COMPLETE
|| info.transitioned_object.name != je.obj_name
crates/ecstore/src/bucket/lifecycle/transition_transaction.rs
+20 -13
View File
@@ -585,7 +585,8 @@ 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, &object, data).await
config_boundary::save_config(api.clone(), &object, data.clone()).await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
pub(crate) async fn load_transition_transaction_record(
@@ -597,8 +598,14 @@ pub(crate) async fn load_transition_transaction_record(
TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)
}
pub(crate) async fn delete_transition_transaction_record(api: Arc<ECStore>, transaction_id: Uuid) -> EcstoreResult<()> {
let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?;
pub(crate) async fn delete_transition_transaction_record(
api: Arc<ECStore>,
transaction: &TransitionTransaction,
) -> 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)?;
api.record_durable_ilm_decommission_terminal(&object, &data).await?;
match config_boundary::delete_config(api, &object).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
Err(err) => Err(err),
@@ -814,7 +821,7 @@ pub async fn finalize_missing_transition_transaction_for_operator(
if probe != TransitionOperatorProbe::Missing {
return Err(TransitionOperatorError::CandidateNotMissing(probe));
}
delete_transition_transaction_record(api, transaction_id)
delete_transition_transaction_record(api, &transaction)
.await
.map_err(TransitionOperatorError::Store)
}
@@ -850,22 +857,22 @@ pub async fn process_transition_transaction_record(
match transaction.state {
TransitionTransactionState::Uploaded => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).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.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).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.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
Err(err) => Err(err),
@@ -873,7 +880,7 @@ pub async fn process_transition_transaction_record(
TransitionTransactionState::LocalCommitStarted => {
match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
@@ -882,7 +889,7 @@ pub async fn process_transition_transaction_record(
}
}
TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
delete_transition_transaction_record(api, transaction.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await,
@@ -908,7 +915,7 @@ async fn recover_unknown_upload_outcome(
.map_err(Error::other)?
{
TransitionCandidateProbe::Missing => {
delete_transition_transaction_record(api, transaction.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionCandidateProbe::UnversionedPresent => {
@@ -926,7 +933,7 @@ async fn recover_unknown_upload_outcome(
)
.await
.map_err(Error::other)?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
delete_transition_transaction_record(api, transaction).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
TransitionCandidateProbe::VersionedPresent(version_id) => {
@@ -959,7 +966,7 @@ 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.transaction_id).await?;
delete_transition_transaction_record(api, &cleanup).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
crates/ecstore/src/config/com.rs
+12 -1
View File
@@ -410,10 +410,21 @@ pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &st
where
S: EcstoreObjectIO,
{
let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await?;
let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
Ok(data)
}
pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
api: Arc<S>,
file: &str,
max_bytes: usize,
) -> Result<(Vec<u8>, ObjectInfo)>
where
S: EcstoreObjectIO,
{
read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
}
/// Read an existing config object without treating an empty payload as absent.
/// Callers that validate their own payload format need to distinguish corruption
/// from `ConfigNotFound`.
crates/ecstore/src/core/pools.rs
+439 -106
View File
@@ -16,7 +16,7 @@ use crate::bucket::replication::replication_state_from_filemeta;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::bucket::{
lifecycle::{
ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord,
DurableIlmRecordCheckpoint, ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord,
bucket_lifecycle_audit::LcEventSrc,
bucket_lifecycle_ops::{
LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule_in, eval_action_from_lifecycle,
@@ -30,8 +30,8 @@ use crate::bucket::{
};
use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
use crate::config::com::{
CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty, read_config_no_lock, save_config,
save_config_with_opts,
CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty,
read_config_limited_preserve_empty_with_metadata, read_config_no_lock, save_config, save_config_with_opts,
};
use crate::data_movement;
use crate::data_movement::backpressure::{self, DataMovementOperation};
@@ -54,7 +54,7 @@ use crate::storage_api_contracts::{
heal::HealOperations as _,
list::ListOperations as _,
namespace::NamespaceLocking as _,
object::{EcstoreObjectIO, ObjectIO as _, ObjectOperations as _},
object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _},
};
use crate::{core::sets::Sets, store::ECStore};
use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
@@ -104,8 +104,9 @@ const DECOMMISSION_TARGET_CAPACITY_OVERHEAD_PERCENT: usize = 30;
const DECOMMISSION_LISTING_MAX_ATTEMPTS: usize = 3;
const DECOMMISSION_LISTING_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
const DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT: &str = "decommission/ilm-receipts";
const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v1";
const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2";
const DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE: usize = 16 * 1024;
const DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS: usize = 3;
/// Background decommission walks must tolerate slow object migrations; the
/// stall timeout is the drive-health bound, not the total listing duration.
const DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
@@ -857,22 +858,24 @@ async fn record_decommission_entry_error(
fn validate_decommission_durable_ilm_copy(
path: &str,
source_record: &ValidatedDurableIlmRecord,
source: &[u8],
target: &[u8],
) -> Result<()> {
validate_durable_ilm_record(path, target).map_err(|err| {
) -> Result<ValidatedDurableIlmRecord> {
let target_record = validate_durable_ilm_record(path, target).map_err(|err| {
Error::other(format!(
"target durable ILM record is invalid at path `{path}` {}: {err}",
source_record.context()
))
})?;
if source != target {
return Err(Error::other(format!(
"target durable ILM record content mismatch at path `{path}` {}",
source_record.context()
)));
}
Ok(())
source_record
.checkpoint
.validate_successor(&target_record.checkpoint)
.map_err(|err| {
Error::other(format!(
"target durable ILM record generation mismatch at path `{path}` {}: {err}",
source_record.context()
))
})?;
Ok(target_record)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -882,17 +885,19 @@ struct DecommissionDurableIlmReceipt {
namespace: String,
id_kind: String,
id: String,
target_content_sha256: String,
checkpoint: DurableIlmRecordCheckpoint,
terminal_checkpoint: Option<DurableIlmRecordCheckpoint>,
}
impl DecommissionDurableIlmReceipt {
fn new(path: &str, record: &ValidatedDurableIlmRecord, target: &[u8]) -> Self {
fn new(path: &str, record: &ValidatedDurableIlmRecord) -> Self {
Self {
source_path: path.to_string(),
namespace: record.namespace.to_string(),
id_kind: record.id_kind.to_string(),
id: record.id.clone(),
target_content_sha256: hex_sha256(target, ToOwned::to_owned),
checkpoint: record.checkpoint.clone(),
terminal_checkpoint: None,
}
}
@@ -915,13 +920,22 @@ impl DecommissionDurableIlmReceipt {
self.source_path
)));
}
if !is_sha256_checksum(&self.target_content_sha256) {
if !is_sha256_checksum(self.checkpoint.content_sha256()) {
return Err(Error::other(format!(
"receipt target checksum is invalid for source path `{}` {}",
self.source_path,
self.context()
)));
}
if let Some(terminal_checkpoint) = &self.terminal_checkpoint {
self.checkpoint.validate_successor(terminal_checkpoint).map_err(|err| {
Error::other(format!(
"receipt terminal checkpoint is invalid for source path `{}` {}: {err}",
self.source_path,
self.context()
))
})?;
}
Ok(())
}
@@ -973,14 +987,80 @@ struct PersistedDecommissionDurableIlmReceipt {
receipt: DecommissionDurableIlmReceipt,
}
fn decommission_durable_ilm_receipt_prefix(cmd_line: &str) -> String {
let pool_key = hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned);
format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{pool_key}/")
#[derive(Debug, Clone, PartialEq, Eq)]
struct DecommissionDurableIlmReceiptLocator {
run_token: String,
source_path: String,
id_kind: String,
id: String,
}
fn decommission_durable_ilm_receipt_path(prefix: &str, source_path: &str) -> String {
let source_key = hex_sha256(source_path.as_bytes(), ToOwned::to_owned);
format!("{prefix}{source_key}.json")
impl DecommissionDurableIlmReceiptLocator {
fn context(&self) -> String {
format!("source path `{}` {} `{}`", self.source_path, self.id_kind, self.id)
}
}
fn decommission_durable_ilm_receipt_run_token(cmd_line: &str) -> String {
hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned)
}
fn decommission_durable_ilm_receipt_run_prefix(run_token: &str) -> String {
format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{run_token}/")
}
fn decommission_durable_ilm_receipt_path(run_token: &str, source_path: &str, id_kind: &str, id: &str) -> String {
format!(
"{}{}/{}/{}.json",
decommission_durable_ilm_receipt_run_prefix(run_token),
source_path,
id_kind,
id
)
}
fn parse_decommission_durable_ilm_receipt_path(path: &str) -> Result<DecommissionDurableIlmReceiptLocator> {
let prefix = format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/");
let suffix = path
.strip_prefix(&prefix)
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has the wrong root")))?;
let (run_token, record_path) = suffix
.split_once('/')
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record path")))?;
let mut parts = record_path.rsplitn(3, '/');
let id = parts
.next()
.and_then(|file| file.strip_suffix(".json"))
.filter(|id| !id.is_empty())
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record id")))?;
let id_kind = parts
.next()
.filter(|id_kind| matches!(*id_kind, "operation_id" | "transaction_id" | "job_id"))
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has an invalid id kind")))?;
let source_path = parts
.next()
.filter(|source_path| !source_path.is_empty())
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its source path")))?;
if !is_sha256_checksum(run_token) {
return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid run token")));
}
match id_kind {
"operation_id" if !is_sha256_checksum(id) => {
return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid operation id")));
}
"transaction_id" | "job_id" if uuid::Uuid::parse_str(id).is_err() => {
return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid UUID")));
}
_ => {}
}
classify_durable_ilm_record(source_path)?
.ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` does not identify a durable ILM source path")))?;
Ok(DecommissionDurableIlmReceiptLocator {
run_token: run_token.to_string(),
source_path: source_path.to_string(),
id_kind: id_kind.to_string(),
id: id.to_string(),
})
}
fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
@@ -4543,7 +4623,7 @@ impl ECStore {
Ok(ret)
}
async fn durable_ilm_receipt_prefix(&self, source_pool_idx: usize) -> Result<String> {
async fn durable_ilm_receipt_run_token(&self, source_pool_idx: usize) -> Result<String> {
let pool_meta = self.pool_meta.read().await;
let cmd_line = pool_meta
.pools
@@ -4551,7 +4631,7 @@ impl ECStore {
.ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))?
.cmd_line
.clone();
Ok(decommission_durable_ilm_receipt_prefix(&cmd_line))
Ok(decommission_durable_ilm_receipt_run_token(&cmd_line))
}
async fn load_decommissioned_durable_ilm_target(
@@ -4560,7 +4640,7 @@ impl ECStore {
path: &str,
max_record_size: usize,
record_context: &str,
) -> Result<(usize, Vec<u8>)> {
) -> Result<Option<(usize, Vec<u8>)>> {
let mut target = None::<(usize, Vec<u8>)>;
let mut first_read_error = None;
for (target_pool_idx, pool) in self.pools.iter().enumerate() {
@@ -4592,40 +4672,62 @@ impl ECStore {
}
}
target.ok_or_else(|| {
first_read_error.unwrap_or_else(|| {
Error::other(format!("target durable ILM record is missing at path `{path}` {record_context}"))
})
})
match (target, first_read_error) {
(Some(target), _) => Ok(Some(target)),
(None, Some(err)) => Err(err),
(None, None) => Ok(None),
}
}
async fn list_decommission_durable_ilm_receipt_paths_in_pool(&self, pool_idx: usize, prefix: &str) -> Result<Vec<String>> {
let pool = self
.pools
.get(pool_idx)
.ok_or_else(|| invalid_decommission_pool_index_error(self.pools.len(), pool_idx))?;
let mut receipts = Vec::new();
let mut continuation = None;
loop {
let page = pool
.clone()
.list_objects_v2(RUSTFS_META_BUCKET, prefix, continuation, None, 1000, false, None, false)
.await
.map_err(|err| {
Error::other(format!(
"failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}"
))
})?;
receipts.extend(page.objects.into_iter().map(|object| object.name));
if !page.is_truncated {
break;
}
continuation = Some(page.next_continuation_token.ok_or_else(|| {
Error::other(format!(
"durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token"
))
})?);
}
Ok(receipts)
}
async fn list_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<Vec<(usize, String)>> {
let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
let prefix = decommission_durable_ilm_receipt_run_prefix(&run_token);
let mut receipts = Vec::new();
for (pool_idx, pool) in self.pools.iter().enumerate() {
for pool_idx in 0..self.pools.len() {
if pool_idx == source_pool_idx {
continue;
}
let mut continuation = None;
loop {
let page = pool
.clone()
.list_objects_v2(RUSTFS_META_BUCKET, &prefix, continuation, None, 1000, false, None, false)
.await
.map_err(|err| {
Error::other(format!(
"failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}"
))
})?;
receipts.extend(page.objects.into_iter().map(|object| (pool_idx, object.name)));
if !page.is_truncated {
break;
for receipt_path in self
.list_decommission_durable_ilm_receipt_paths_in_pool(pool_idx, &prefix)
.await?
{
let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?;
if locator.run_token != run_token {
return Err(Error::other(format!(
"durable ILM receipt path `{receipt_path}` has an unexpected run token"
)));
}
continuation = Some(page.next_continuation_token.ok_or_else(|| {
Error::other(format!(
"durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token"
))
})?);
receipts.push((pool_idx, receipt_path));
}
}
Ok(receipts)
@@ -4637,8 +4739,8 @@ impl ECStore {
target_pool_idx: usize,
receipt: &DecommissionDurableIlmReceipt,
) -> Result<()> {
let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
let receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path);
let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
let receipt_path = decommission_durable_ilm_receipt_path(&run_token, &receipt.source_path, &receipt.id_kind, &receipt.id);
let encoded = receipt.encode().map_err(|err| {
Error::other(format!(
"failed to encode durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}",
@@ -4657,36 +4759,57 @@ impl ECStore {
})
}
fn validate_decommission_durable_ilm_receipt_locator(
receipt_path: &str,
locator: &DecommissionDurableIlmReceiptLocator,
receipt: &DecommissionDurableIlmReceipt,
) -> Result<()> {
if locator.source_path != receipt.source_path || locator.id_kind != receipt.id_kind || locator.id != receipt.id {
return Err(Error::other(format!(
"durable ILM decommission receipt path `{receipt_path}` identity {} does not match receipt {}",
locator.context(),
receipt.context()
)));
}
Ok(())
}
async fn read_decommission_durable_ilm_receipt(
&self,
receipt_pool_idx: usize,
receipt_path: &str,
) -> Result<DecommissionDurableIlmReceipt> {
let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?;
let data = read_config_limited_preserve_empty(
self.pools[receipt_pool_idx].clone(),
receipt_path,
DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
)
.await
.map_err(|err| {
Error::other(format!(
"failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx} for {}: {err}",
locator.context()
))
})?;
let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| {
Error::other(format!(
"durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} for {} is invalid: {err}",
locator.context()
))
})?;
Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?;
Ok(receipt)
}
async fn verify_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> {
let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
for (receipt_pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? {
let data = read_config_limited_preserve_empty(
self.pools[receipt_pool_idx].clone(),
&receipt_path,
DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
)
.await
.map_err(|err| {
Error::other(format!(
"failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx}: {err}"
))
})?;
let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| {
Error::other(format!(
"durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} is invalid: {err}"
))
})?;
let expected_receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path);
if receipt_path != expected_receipt_path {
return Err(Error::other(format!(
"durable ILM decommission receipt path `{receipt_path}` does not match source path `{}` {}",
receipt.source_path,
receipt.context()
)));
}
let receipt = self
.read_decommission_durable_ilm_receipt(receipt_pool_idx, &receipt_path)
.await?;
let namespace = classify_durable_ilm_record(&receipt.source_path)?
.ok_or_else(|| Error::other(format!("path `{}` is not a durable ILM record", receipt.source_path)))?;
let (_, target) = self
let target = self
.load_decommissioned_durable_ilm_target(
source_pool_idx,
&receipt.source_path,
@@ -4694,28 +4817,50 @@ impl ECStore {
&receipt.context(),
)
.await?;
let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| {
Error::other(format!(
"target durable ILM record is invalid at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
if target_record.namespace != receipt.namespace
|| target_record.id_kind != receipt.id_kind
|| target_record.id != receipt.id
{
if let Some((_, target)) = target {
let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| {
Error::other(format!(
"target durable ILM record is invalid at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
let identity_matches = target_record.namespace == receipt.namespace
&& target_record.id_kind == receipt.id_kind
&& target_record.id == receipt.id;
let reused_manual_scope = receipt.terminal_checkpoint.is_some()
&& matches!(
(&receipt.checkpoint, &target_record.checkpoint),
(
DurableIlmRecordCheckpoint::ManualTransitionScope { .. },
DurableIlmRecordCheckpoint::ManualTransitionScope { .. }
)
);
if !identity_matches && !reused_manual_scope {
return Err(Error::other(format!(
"target durable ILM record identity mismatch at path `{}` {}; decoded {}",
receipt.source_path,
receipt.context(),
target_record.context()
)));
}
if identity_matches {
receipt
.terminal_checkpoint
.as_ref()
.unwrap_or(&receipt.checkpoint)
.validate_successor(&target_record.checkpoint)
.map_err(|err| {
Error::other(format!(
"target durable ILM record generation mismatch at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
}
} else if receipt.terminal_checkpoint.is_none() {
return Err(Error::other(format!(
"target durable ILM record identity mismatch at path `{}` {}; decoded {}",
receipt.source_path,
receipt.context(),
target_record.context()
)));
}
let target_content_sha256 = hex_sha256(&target, ToOwned::to_owned);
if target_content_sha256 != receipt.target_content_sha256 {
return Err(Error::other(format!(
"target durable ILM record content mismatch at path `{}` {}",
"target durable ILM record is missing at path `{}` {} without a recovery terminal checkpoint",
receipt.source_path,
receipt.context()
)));
@@ -4724,6 +4869,180 @@ impl ECStore {
Ok(())
}
async fn advance_durable_ilm_decommission_receipt(
&self,
pool_idx: usize,
receipt_path: &str,
record: &ValidatedDurableIlmRecord,
terminal: bool,
) -> Result<bool> {
let stage = if terminal { "terminal" } else { "progress" };
let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?;
let mut attempt = 1;
loop {
let (receipt_data, metadata) = match read_config_limited_preserve_empty_with_metadata(
self.pools[pool_idx].clone(),
receipt_path,
DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
)
.await
{
Ok(receipt) => receipt,
Err(err)
if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound)
|| is_err_object_not_found(&err)
|| is_err_version_not_found(&err) =>
{
return Ok(false);
}
Err(err) => {
return Err(Error::other(format!(
"failed to read durable ILM decommission receipt `{receipt_path}` from pool {pool_idx} for {}: {err}",
locator.context()
)));
}
};
let mut receipt = DecommissionDurableIlmReceipt::decode(&receipt_data).map_err(|err| {
Error::other(format!(
"durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} for {} is invalid: {err}",
locator.context()
))
})?;
Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?;
receipt.checkpoint.validate_successor(&record.checkpoint).map_err(|err| {
Error::other(format!(
"{stage} durable ILM record generation mismatch at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
if terminal {
if let Some(existing) = &receipt.terminal_checkpoint {
if existing == &record.checkpoint || record.checkpoint.validate_successor(existing).is_ok() {
return Ok(true);
}
existing.validate_successor(&record.checkpoint).map_err(|err| {
Error::other(format!(
"terminal durable ILM record checkpoint conflicts at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
}
receipt.terminal_checkpoint = Some(record.checkpoint.clone());
} else {
if let Some(existing) = &receipt.terminal_checkpoint {
if existing == &record.checkpoint {
return Ok(true);
}
existing.validate_successor(&record.checkpoint).map_err(|err| {
Error::other(format!(
"progress durable ILM record conflicts with terminal checkpoint at path `{}` {}: {err}",
receipt.source_path,
receipt.context()
))
})?;
receipt.terminal_checkpoint = None;
}
if receipt.checkpoint == record.checkpoint {
return Ok(true);
}
receipt.checkpoint = record.checkpoint.clone();
}
let etag = metadata.etag.filter(|etag| !etag.trim().is_empty()).ok_or_else(|| {
Error::other(format!(
"durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} is missing an ETag"
))
})?;
let encoded = receipt.encode()?;
match save_config_with_opts(
self.pools[pool_idx].clone(),
receipt_path,
encoded,
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
if_match: Some(etag),
..Default::default()
}),
..Default::default()
},
)
.await
{
Ok(()) => return Ok(true),
Err(Error::PreconditionFailed) if attempt < DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS => {
attempt += 1;
continue;
}
Err(Error::PreconditionFailed) => {
return Err(Error::other(format!(
"failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {} after concurrent updates",
locator.context()
)));
}
Err(err) => {
return Err(Error::other(format!(
"failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {}: {err}",
locator.context()
)));
}
}
}
}
async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<()> {
let active_runs = {
let pool_meta = self.pool_meta.read().await;
pool_meta
.pools
.iter()
.enumerate()
.filter(|(_, pool)| {
pool.decommission
.as_ref()
.is_some_and(|info| info.has_decommission_state() && !info.complete)
})
.map(|(pool_idx, pool)| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line)))
.collect::<Vec<_>>()
};
if active_runs.is_empty() {
return Ok(());
}
let stage = if terminal { "terminal" } else { "progress" };
let record = validate_durable_ilm_record(path, data)
.map_err(|err| Error::other(format!("{stage} durable ILM record is invalid at path `{path}`: {err}")))?;
for (source_pool_idx, run_token) in active_runs {
let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, record.id_kind, &record.id);
let mut receipt_found = false;
for pool_idx in 0..self.pools.len() {
if pool_idx != source_pool_idx {
receipt_found |= self
.advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal)
.await?;
}
}
if terminal && !receipt_found {
return Err(Error::other(format!(
"terminal durable ILM record at path `{path}` {} is retained until its decommission receipt is committed",
record.context()
)));
}
}
Ok(())
}
pub(crate) async fn record_durable_ilm_decommission_progress(&self, path: &str, data: &[u8]) -> Result<()> {
self.advance_durable_ilm_decommission_receipts(path, data, false).await
}
pub(crate) async fn record_durable_ilm_decommission_terminal(&self, path: &str, data: &[u8]) -> Result<()> {
self.advance_durable_ilm_decommission_receipts(path, data, true).await
}
async fn cleanup_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> {
for (pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? {
match delete_config(self.pools[pool_idx].clone(), &receipt_path).await {
@@ -4759,9 +5078,15 @@ impl ECStore {
.map_err(|err| Error::other(format!("source durable ILM record is invalid at path `{path}`: {err}")))?;
let (target_pool_idx, target) = self
.load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context())
.await?;
validate_decommission_durable_ilm_copy(path, &source_record, &source, &target)?;
let receipt = DecommissionDurableIlmReceipt::new(path, &source_record, &target);
.await?
.ok_or_else(|| {
Error::other(format!(
"target durable ILM record is missing at path `{path}` {}",
source_record.context()
))
})?;
let target_record = validate_decommission_durable_ilm_copy(path, &source_record, &target)?;
let receipt = DecommissionDurableIlmReceipt::new(path, &target_record);
self.persist_decommission_durable_ilm_receipt(source_pool_idx, target_pool_idx, &receipt)
.await?;
@@ -4800,6 +5125,14 @@ impl ECStore {
Ok(self.list_decommission_durable_ilm_receipts(source_pool_idx).await?.len())
}
#[cfg(test)]
pub(crate) async fn decommission_durable_ilm_receipt_paths_for_test(
&self,
source_pool_idx: usize,
) -> Result<Vec<(usize, String)>> {
self.list_decommission_durable_ilm_receipts(source_pool_idx).await
}
#[cfg(test)]
pub(crate) async fn cleanup_decommission_durable_ilm_receipts_for_test(&self, source_pool_idx: usize) -> Result<()> {
self.cleanup_decommission_durable_ilm_receipts(source_pool_idx).await
crates/ecstore/src/set_disk/ops/object.rs
+7 -2
View File
@@ -37,7 +37,7 @@ use crate::bucket::lifecycle::{
transition_transaction::{
TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
save_transition_transaction_record,
load_transition_transaction_record, save_transition_transaction_record,
},
};
use crate::bucket::quota::reservation;
@@ -4246,7 +4246,12 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction,
async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
if let Some(api) = api {
return delete_transition_transaction_record(api.clone(), transaction_id).await;
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;
}
Ok(())
}
crates/ecstore/src/store/init.rs
+183 -34
View File
@@ -3055,7 +3055,7 @@ mod tests {
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-decommission", &[4, 4])).await;
let tier_name = "DECOMMISSION-ILM";
let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
let backend = register_transition_reconcile_test_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")
@@ -3073,7 +3073,7 @@ mod tests {
let tier_path = tier_delete_journal_object_name(&tier_entry);
let tier_bytes = encode_tier_delete_journal_entry(&tier_entry).expect("tier journal should encode");
let transaction = TransitionTransaction::new(TransitionTransactionInit {
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(),
@@ -3093,6 +3093,10 @@ mod tests {
not_after_unix_nanos: 1_780_000_000_000_000_000,
})
.expect("transition transaction should build");
let regressed_transaction = transaction.clone();
transaction
.advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
.expect("transition transaction should advance before migration");
let transaction_path = transition_transaction_record_object_name(transaction.transaction_id)
.expect("transition transaction path should build");
let transaction_bytes = transaction.encode().expect("transition transaction should encode");
@@ -3212,6 +3216,25 @@ mod tests {
*expected
);
}
assert_eq!(
store
.decommission_durable_ilm_receipt_count_for_test(0)
.await
.expect("no receipt should exist before the final sweep"),
0
);
let isolated_tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
.await
.expect("tier recovery should retain a terminal record until its receipt is committed");
assert!(isolated_tier_stats.scanned >= 1);
assert_eq!(isolated_tier_stats.deleted, 0);
assert!(isolated_tier_stats.failed >= 1);
assert_eq!(
com::read_config(store.pools[1].clone(), &tier_path)
.await
.expect("receipt isolation must retain the target tier journal"),
tier_bytes
);
com::delete_config(store.pools[1].clone(), &manual_job_path)
.await
@@ -3404,6 +3427,161 @@ mod tests {
.await
.expect("post-crash target transaction should restore");
let mut wrong_manual_job = manual_job.clone();
wrong_manual_job.job_id = uuid::Uuid::new_v4();
com::save_config(
store.pools[1].clone(),
&manual_job_path,
wrong_manual_job.encode().expect("wrong-id job should encode"),
)
.await
.expect("post-crash target manual job should accept the wrong-id fixture");
let wrong_id_after_crash = store
.complete_decommission(0)
.await
.expect_err("completion must reject a target record with the wrong id")
.to_string();
assert!(wrong_id_after_crash.contains(&manual_job_path));
assert!(wrong_id_after_crash.contains(&manual_job_id.to_string()));
com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone())
.await
.expect("post-crash target manual job should restore after the wrong-id check");
com::save_config(
store.pools[1].clone(),
&transaction_path,
regressed_transaction
.encode()
.expect("regressed transition transaction should encode"),
)
.await
.expect("post-crash target transaction should accept the regression fixture");
let regression_after_crash = store
.complete_decommission(0)
.await
.expect_err("completion must reject a lower transition transaction revision")
.to_string();
assert!(regression_after_crash.contains("generation mismatch"));
assert!(regression_after_crash.contains(&transaction_path));
assert!(regression_after_crash.contains(&transaction.transaction_id.to_string()));
com::save_config(store.pools[1].clone(), &transaction_path, transaction_bytes.clone())
.await
.expect("post-crash target transaction should restore after the regression check");
let (manual_task_receipt_pool, manual_task_receipt_path) = store
.decommission_durable_ilm_receipt_paths_for_test(0)
.await
.expect("durable ILM receipt paths should be listable")
.into_iter()
.find(|(_, path)| path.contains(&manual_task_path))
.expect("manual task receipt should retain its reversible source path");
let manual_task_receipt_bytes =
com::read_config(store.pools[manual_task_receipt_pool].clone(), &manual_task_receipt_path)
.await
.expect("manual task receipt should be readable before corruption");
com::save_config(
store.pools[manual_task_receipt_pool].clone(),
&manual_task_receipt_path,
b"{corrupt".to_vec(),
)
.await
.expect("manual task receipt should corrupt deterministically");
let corrupt_receipt = store
.complete_decommission(0)
.await
.expect_err("completion must fail closed on a corrupt receipt")
.to_string();
assert!(corrupt_receipt.contains(&manual_task_path));
assert!(corrupt_receipt.contains(&manual_job_id.to_string()));
com::save_config(
store.pools[manual_task_receipt_pool].clone(),
&manual_task_receipt_path,
manual_task_receipt_bytes,
)
.await
.expect("manual task receipt should restore after the corruption check");
let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
.await
.expect("tier journal recovery should consume the migrated record before completion");
assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0));
assert!(matches!(com::read_config(store.clone(), &tier_path).await, Err(Error::ConfigNotFound)));
let recovered_transition_version = "recovered-transition-version".to_string();
backend
.set_transition_candidate_probe_override(Some(TransitionCandidateProbe::VersionedPresent(
recovered_transition_version.clone(),
)))
.await;
let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("transition recovery should advance and consume the migrated transaction before completion");
backend.set_transition_candidate_probe_override(None).await;
assert_eq!(
(
transaction_stats.scanned,
transaction_stats.recovered,
transaction_stats.retained,
transaction_stats.failed,
),
(1, 1, 0, 0)
);
assert!(matches!(
com::read_config(store.clone(), &transaction_path).await,
Err(Error::ConfigNotFound)
));
com::save_config(
store.pools[1].clone(),
&transaction_path,
regressed_transaction
.encode()
.expect("post-terminal transition rollback should encode"),
)
.await
.expect("target should accept the post-terminal rollback fixture");
let post_terminal_regression = store
.complete_decommission(0)
.await
.expect_err("terminal proof must not mask a lower transition revision")
.to_string();
assert!(post_terminal_regression.contains("generation mismatch"));
assert!(post_terminal_regression.contains(&transaction_path));
assert!(post_terminal_regression.contains(&transaction.transaction_id.to_string()));
com::delete_config(store.pools[1].clone(), &transaction_path)
.await
.expect("post-terminal rollback fixture should be removed");
let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None)
.await
.expect("manual recovery should advance the migrated job and consume its scope before completion");
assert_eq!(
(manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,),
(1, 1, 0, 0)
);
assert!(matches!(
com::read_config(store.clone(), &manual_scope_path).await,
Err(Error::ConfigNotFound)
));
let recovered_manual_job_bytes = com::read_config(store.pools[1].clone(), &manual_job_path)
.await
.expect("manual recovery should retain the advanced job record");
assert_ne!(recovered_manual_job_bytes, manual_job_bytes);
com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone())
.await
.expect("target manual job should accept the rollback fixture");
let manual_regression = store
.complete_decommission(0)
.await
.expect_err("completion must reject a manual job generation rollback")
.to_string();
assert!(manual_regression.contains("generation mismatch"));
assert!(manual_regression.contains(&manual_job_path));
assert!(manual_regression.contains(&manual_job_id.to_string()));
com::save_config(store.pools[1].clone(), &manual_job_path, recovered_manual_job_bytes)
.await
.expect("target manual job should restore its recovered generation");
store
.complete_decommission(0)
.await
@@ -3426,38 +3604,9 @@ mod tests {
.cleanup_decommission_durable_ilm_receipts_for_test(0)
.await
.expect("receipt cleanup should be idempotent");
let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
.await
.expect("tier journal recovery should consume the migrated record");
assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0));
assert_eq!(
backend.remove_versions().await,
vec![(tier_entry.obj_name.clone(), tier_entry.version_id.clone())]
);
let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None)
.await
.expect("transition transaction recovery should read the migrated record");
assert_eq!(
(
transaction_stats.scanned,
transaction_stats.recovered,
transaction_stats.retained,
transaction_stats.failed,
),
(1, 0, 1, 0)
);
let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None)
.await
.expect("manual recovery should reconcile migrated job, scope, task, and result records");
assert_eq!(
(manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,),
(1, 1, 0, 0)
);
assert!(matches!(
com::read_config(store.clone(), &manual_scope_path).await,
Err(Error::ConfigNotFound)
));
let removed_versions = backend.remove_versions().await;
assert!(removed_versions.contains(&(tier_entry.obj_name.clone(), tier_entry.version_id.clone())));
assert!(removed_versions.contains(&(transaction.remote_object.clone(), recovered_transition_version)));
}
#[cfg(feature = "test-util")]