gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 13:16:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): preserve ILM state during pool decommission (#6369)

Browse Source 
* fix(ecstore): migrate ILM metadata during decommission

* fix(ecstore): verify ILM metadata before decommission

* fix(ecstore): track ILM recovery across decommission

* fix(ecstore): close ILM receipt recovery gaps

* fix(ecstore): anchor decommission ILM receipts

* fix(ecstore): re-export durable ILM checkpoint

* fix(ecstore): avoid terminal receipt shadowing

* fix(ecstore): harden durable ILM cursor receipts

* fix(ecstore): repair durable ILM receipt recovery

* test(ecstore): cover durable ILM recovery boundaries

* test(ecstore): serialize multi-source ILM recovery

* test(ecstore): compile multi-source ILM recovery

* test(ecstore): isolate durable ILM scenario stack

* fix(ecstore): preserve active ILM source journals

* fix(ecstore): distinguish active ILM target cleanup

* fix(ecstore): restore decommission test imports

* fix(ecstore): drop removed decommission test import

* fix(ecstore): remove duplicate decommission error helper

* fix(ecstore): fence final decommission sweep

* test(ecstore): cover final sweep cancel fence

* fix(ecstore): fence decommission cancellation

* fix(ecstore): remove redundant clone in test

* fix(ecstore): keep manual transition progress compatible

* fix(ecstore): restore decommission worker wrapper

* fix(ecstore): restore decommission compile contracts

* test(ecstore): adapt reload worker canceler
This commit is contained in:
Zhengchao An
2026-08-23 16:43:11 +08:00
committed by GitHub
parent 450ec7f66a
commit 34bbc1adb3
12 changed files with 4272 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/nextest.toml
+10
View File
@@ -78,6 +78,12 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
test-group = 'ecstore-serial-flaky'
# The durable ILM decommission regressions build isolated multi-pool stores and
# deliberately take source or target disks offline while checking fencing.
[[profile.default.overrides]]
filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
test-group = 'ecstore-serial-flaky'
# Serialize the bucket-incarnation / lifecycle-fence tests. They drive
# init_bucket_metadata_sys and bucket_metadata_sys_of, i.e. process-global
# OnceLock state that serial_test's #[serial] cannot protect across nextest's
@@ -190,6 +196,10 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
test-group = 'ecstore-serial-flaky'
[[profile.ci.overrides]]
filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
test-group = 'ecstore-serial-flaky'
# Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile
# too (see the matching default-profile override near the top). No retries.
[[profile.ci.overrides]]
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+70 -3
View File
@@ -3739,17 +3739,55 @@ impl ManualTransitionRunReport {
}
pub fn merge_scan_report_preserving_worker(&mut self, scan_report: &ManualTransitionRunReport) {
let previous = self.clone();
let resumed_after_checkpoint = previous.continuation_token.is_some() && scan_report.scanned < previous.scanned;
let mut tier_failure_by_reason = self.tier_failure_by_reason.clone();
for (reason, count) in &scan_report.tier_failure_by_reason {
let current = tier_failure_by_reason.get(reason).copied().unwrap_or_default();
tier_failure_by_reason.insert(*reason, current.max(*count));
let merged = if resumed_after_checkpoint {
current.saturating_add(*count)
} else {
current.max(*count)
};
tier_failure_by_reason.insert(*reason, merged);
}
let transition_completed = self.transition_completed;
let transition_failed = self.transition_failed;
*self = scan_report.clone();
if resumed_after_checkpoint {
self.scanned = previous.scanned.saturating_add(scan_report.scanned);
self.eligible = previous.eligible.saturating_add(scan_report.eligible);
self.enqueued = previous.enqueued.saturating_add(scan_report.enqueued);
self.dry_run_eligible = previous.dry_run_eligible.saturating_add(scan_report.dry_run_eligible);
self.skipped_not_transition = previous
.skipped_not_transition
.saturating_add(scan_report.skipped_not_transition);
self.skipped_tier = previous.skipped_tier.saturating_add(scan_report.skipped_tier);
self.skipped_delete_marker = previous
.skipped_delete_marker
.saturating_add(scan_report.skipped_delete_marker);
self.skipped_directory = previous.skipped_directory.saturating_add(scan_report.skipped_directory);
self.skipped_replication = previous.skipped_replication.saturating_add(scan_report.skipped_replication);
self.skipped_already_transitioned = previous
.skipped_already_transitioned
.saturating_add(scan_report.skipped_already_transitioned);
self.skipped_already_in_flight = previous
.skipped_already_in_flight
.saturating_add(scan_report.skipped_already_in_flight);
self.skipped_queue_full = previous.skipped_queue_full.saturating_add(scan_report.skipped_queue_full);
self.skipped_queue_closed = previous.skipped_queue_closed.saturating_add(scan_report.skipped_queue_closed);
self.skipped_queue_timeout = previous
.skipped_queue_timeout
.saturating_add(scan_report.skipped_queue_timeout);
self.tier_failure = previous.tier_failure.saturating_add(scan_report.tier_failure);
}
self.lifecycle_config_found = previous.lifecycle_config_found || scan_report.lifecycle_config_found;
self.truncated_by_limit = previous.truncated_by_limit || scan_report.truncated_by_limit;
self.truncated_by_duration = previous.truncated_by_duration || scan_report.truncated_by_duration;
self.cancelled = previous.cancelled || scan_report.cancelled;
self.transition_completed = transition_completed;
self.transition_failed = transition_failed;
self.tier_failure = scan_report.tier_failure.saturating_add(transition_failed);
self.tier_failure = self.tier_failure.saturating_add(transition_failed);
self.tier_failure_by_reason = tier_failure_by_reason;
}
@@ -3765,7 +3803,10 @@ struct ManualTransitionContinuationToken {
version_marker: Option<String>,
}
fn encode_manual_transition_continuation_token(marker: Option<String>, version_marker: Option<String>) -> Option<String> {
pub(super) fn encode_manual_transition_continuation_token(
marker: Option<String>,
version_marker: Option<String>,
) -> Option<String> {
if marker.is_none() && version_marker.is_none() {
return None;
}
@@ -9136,6 +9177,7 @@ mod tests {
assert_eq!(loaded.report.scanned, 37);
assert_eq!(loaded.report.eligible, 11);
assert_eq!(loaded.report.enqueued, 5);
assert_eq!(loaded.cursor_revision, Some(37));
assert!(loaded.lease_expires_at_unix_nanos > 0);
let token = loaded
.report
@@ -9154,6 +9196,30 @@ mod tests {
assert_eq!(admission.lease_id, loaded.lease_id);
assert_eq!(admission.lease_expires_at_unix_nanos, loaded.lease_expires_at_unix_nanos);
let mut same_marker_report = report.clone();
same_marker_report.scanned += 1;
persist_manual_transition_page_checkpoint(
&checkpoint_options,
&same_marker_report,
Some("logs/page-end".to_string()),
Some("opaque-next-version".to_string()),
)
.await
.expect("same-marker version checkpoint should persist through the durable progress sink");
let same_marker_checkpointed = load_manual_transition_job_record(ecstore.clone(), job_id)
.await
.expect("same-marker version checkpoint should reload");
assert_eq!(same_marker_checkpointed.cursor_revision, Some(38));
let (_, version_marker) = decode_manual_transition_continuation_token(
same_marker_checkpointed
.report
.continuation_token
.as_deref()
.expect("same-marker version checkpoint should persist a cursor"),
)
.expect("same-marker version cursor should decode");
assert_eq!(version_marker.as_deref(), Some("opaque-next-version"));
create_test_bucket(&ecstore, &bucket).await;
let lifecycle_xml = format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -9210,6 +9276,7 @@ mod tests {
assert_eq!(checkpointed.report.scanned, 1000);
assert_eq!(checkpointed.report.eligible, 1000);
assert_eq!(checkpointed.report.dry_run_eligible, 1000);
assert_eq!(checkpointed.cursor_revision, Some(1000));
let token = checkpointed
.report
.continuation_token
crates/ecstore/src/bucket/lifecycle/durable_namespace.rs
+1161
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs
+217 -25
View File
@@ -20,10 +20,16 @@ use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use uuid::Uuid;
#[cfg(test)]
use crate::bucket::lifecycle::bucket_lifecycle_ops::encode_manual_transition_continuation_token;
use crate::bucket::lifecycle::bucket_lifecycle_ops::{
ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport,
};
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::{
MANUAL_TRANSITION_JOB_NAMESPACE, MANUAL_TRANSITION_SCOPE_NAMESPACE, MANUAL_TRANSITION_TASK_NAMESPACE,
MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
};
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::{Error, Result as EcstoreResult};
use crate::object_api::ObjectOptions;
@@ -34,10 +40,10 @@ use crate::store::ECStore;
pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1";
pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1";
pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1";
pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs";
pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes";
pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks";
pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results";
pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = MANUAL_TRANSITION_JOB_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = MANUAL_TRANSITION_SCOPE_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_TASK_PREFIX: &str = MANUAL_TRANSITION_TASK_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE.prefix;
pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024;
pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024;
pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024;
@@ -195,6 +201,8 @@ pub struct ManualTransitionJobRecord {
pub updated_at_unix_nanos: i128,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub completed_at_unix_nanos: Option<i128>,
#[serde(default, skip_serializing)]
pub cursor_revision: Option<u64>,
pub report: ManualTransitionRunReport,
pub queue_snapshot: ManualTransitionQueueSnapshot,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -224,6 +232,7 @@ impl ManualTransitionJobRecord {
created_at_unix_nanos: now,
updated_at_unix_nanos: now,
completed_at_unix_nanos: None,
cursor_revision: None,
report: ManualTransitionRunReport {
bucket: bucket.to_string(),
prefix: options.prefix.clone(),
@@ -238,7 +247,7 @@ impl ManualTransitionJobRecord {
pub fn complete(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
self.scan_completed = true;
self.report.merge_scan_report_preserving_worker(&report);
self.merge_scan_report(&report);
self.queue_snapshot = queue_snapshot;
self.error = None;
self.mark_terminal_if_worker_drained();
@@ -316,7 +325,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();
}
@@ -359,14 +368,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) {
@@ -380,7 +389,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();
}
}
@@ -398,7 +407,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;
}
@@ -438,11 +447,16 @@ impl ManualTransitionJobRecord {
pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
if self.state == ManualTransitionJobState::Running {
self.report.merge_scan_report_preserving_worker(&report);
self.merge_scan_report(&report);
self.renew_lease(queue_snapshot);
}
}
fn merge_scan_report(&mut self, report: &ManualTransitionRunReport) {
self.report.merge_scan_report_preserving_worker(report);
self.cursor_revision = manual_transition_cursor_revision(&self.report);
}
pub fn mark_unknown_if_unowned(&mut self) {
if self.state == ManualTransitionJobState::Running {
self.state = ManualTransitionJobState::Unknown;
@@ -463,9 +477,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) {
@@ -543,6 +561,7 @@ impl ManualTransitionJobRecord {
if job.state == ManualTransitionJobState::Cancelled && job.cancel_requested {
job.report.cancelled = true;
}
job.cursor_revision = manual_transition_cursor_revision(&job.report);
job.validate()?;
Ok(job)
}
@@ -1109,7 +1128,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> {
@@ -1142,9 +1162,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 {
@@ -1154,7 +1174,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.
@@ -1592,9 +1613,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 {
@@ -1604,7 +1625,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(
@@ -1642,9 +1664,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 {
@@ -1660,7 +1682,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(
@@ -1953,13 +1976,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),
@@ -1971,6 +1996,11 @@ fn manual_transition_job_store_error(err: ManualTransitionJobError) -> Error {
Error::other(err)
}
fn manual_transition_cursor_revision(report: &ManualTransitionRunReport) -> Option<u64> {
report.continuation_token.as_ref()?;
(report.scanned > 0).then_some(report.scanned)
}
pub fn manual_transition_scope_admission_lease_expired(admission: &ManualTransitionScopeAdmission) -> bool {
OffsetDateTime::now_utc().unix_timestamp_nanos() > admission.lease_expires_at_unix_nanos
}
@@ -2210,6 +2240,76 @@ mod tests {
);
}
#[test]
fn manual_transition_job_scan_progress_accumulates_resumed_checkpoint_counters() {
let options = ManualTransitionRunOptions::default();
let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
let first_token =
encode_manual_transition_continuation_token(Some("logs/page-a".to_string()), Some("version-a".to_string()));
record.update_running_progress(
ManualTransitionRunReport {
bucket: "bucket".to_string(),
lifecycle_config_found: true,
scanned: 1000,
eligible: 800,
enqueued: 50,
dry_run_eligible: 10,
skipped_not_transition: 2,
skipped_tier: 3,
skipped_delete_marker: 4,
skipped_directory: 5,
skipped_replication: 6,
skipped_already_transitioned: 7,
skipped_already_in_flight: 8,
skipped_queue_full: 9,
skipped_queue_closed: 10,
skipped_queue_timeout: 11,
tier_failure: 12,
truncated_by_duration: true,
continuation_token: first_token,
..Default::default()
},
ManualTransitionQueueSnapshot::default(),
);
let next_token =
encode_manual_transition_continuation_token(Some("logs/page-b".to_string()), Some("version-b".to_string()));
record.update_running_progress(
ManualTransitionRunReport {
bucket: "bucket".to_string(),
scanned: 3,
eligible: 2,
enqueued: 1,
dry_run_eligible: 1,
skipped_not_transition: 1,
tier_failure: 1,
continuation_token: next_token.clone(),
..Default::default()
},
ManualTransitionQueueSnapshot::default(),
);
assert_eq!(record.report.scanned, 1003);
assert_eq!(record.report.eligible, 802);
assert_eq!(record.report.enqueued, 51);
assert_eq!(record.report.dry_run_eligible, 11);
assert_eq!(record.report.skipped_not_transition, 3);
assert_eq!(record.report.skipped_tier, 3);
assert_eq!(record.report.skipped_delete_marker, 4);
assert_eq!(record.report.skipped_directory, 5);
assert_eq!(record.report.skipped_replication, 6);
assert_eq!(record.report.skipped_already_transitioned, 7);
assert_eq!(record.report.skipped_already_in_flight, 8);
assert_eq!(record.report.skipped_queue_full, 9);
assert_eq!(record.report.skipped_queue_closed, 10);
assert_eq!(record.report.skipped_queue_timeout, 11);
assert_eq!(record.report.tier_failure, 13);
assert!(record.report.lifecycle_config_found);
assert!(record.report.truncated_by_duration);
assert_eq!(record.report.continuation_token, next_token);
assert_eq!(record.cursor_revision, Some(1003));
}
#[test]
fn manual_transition_job_apply_worker_result_counts_preserves_existing_failure_reasons() {
let options = ManualTransitionRunOptions::default();
@@ -2577,6 +2677,98 @@ mod tests {
assert!(decoded.report.tier_failure_by_reason.is_empty());
}
#[test]
fn manual_transition_job_record_derives_revision_from_legacy_cursor() {
let options = ManualTransitionRunOptions::default();
let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
let continuation_token =
encode_manual_transition_continuation_token(Some("logs/page-a".to_string()), Some("version-a".to_string()));
record.update_running_progress(
ManualTransitionRunReport {
bucket: "bucket".to_string(),
scanned: 9,
continuation_token,
..Default::default()
},
ManualTransitionQueueSnapshot::default(),
);
let encoded = record.encode().expect("job record should encode");
let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded job should be json");
value["job"]
.as_object_mut()
.expect("job should be object")
.remove("cursor_revision");
let record_bytes = serde_json::to_vec(&value["job"]).expect("legacy job should encode");
value["content_sha256"] = serde_json::Value::String(hex_sha256(&record_bytes, ToOwned::to_owned));
let legacy = serde_json::to_vec(&value).expect("legacy envelope should encode");
let decoded = ManualTransitionJobRecord::decode(record.job_id, &legacy).expect("legacy job should decode");
assert_eq!(decoded.cursor_revision, Some(9));
}
#[test]
fn manual_transition_job_record_omits_cursor_revision_for_old_readers() {
#[allow(dead_code)]
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyPersistedManualTransitionJobRecord {
schema: String,
content_sha256: String,
job: LegacyManualTransitionJobRecord,
}
#[allow(dead_code)]
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyManualTransitionJobRecord {
job_id: Uuid,
scope_key: String,
bucket: String,
prefix: String,
tier: Option<String>,
dry_run: bool,
max_objects: Option<u64>,
max_duration: Option<std::time::Duration>,
owner_id: String,
lease_id: Uuid,
lease_expires_at_unix_nanos: i128,
state: ManualTransitionJobState,
scan_completed: bool,
cancel_requested: bool,
created_at_unix_nanos: i128,
updated_at_unix_nanos: i128,
completed_at_unix_nanos: Option<i128>,
report: ManualTransitionRunReport,
queue_snapshot: ManualTransitionQueueSnapshot,
error: Option<String>,
}
let options = ManualTransitionRunOptions::default();
let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
let continuation_token =
encode_manual_transition_continuation_token(Some("logs/page-a".to_string()), Some("version-a".to_string()));
record.update_running_progress(
ManualTransitionRunReport {
bucket: "bucket".to_string(),
scanned: 7,
continuation_token: continuation_token.clone(),
..Default::default()
},
ManualTransitionQueueSnapshot::default(),
);
assert_eq!(record.cursor_revision, Some(7));
let encoded = record.encode().expect("job record should encode");
let value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded job should be json");
assert!(value["job"].get("cursor_revision").is_none());
let legacy: LegacyPersistedManualTransitionJobRecord =
serde_json::from_slice(&encoded).expect("old reader should accept new job record");
assert_eq!(legacy.job.job_id, record.job_id);
assert_eq!(legacy.job.report.continuation_token, continuation_token);
}
#[test]
fn manual_transition_job_record_rejects_unknown_report_fields() {
let options = ManualTransitionRunOptions::default();
crates/ecstore/src/bucket/lifecycle/mod.rs
+6
View File
@@ -16,6 +16,7 @@ pub mod bucket_lifecycle_audit;
pub mod bucket_lifecycle_ops;
mod config_boundary;
pub mod core;
mod durable_namespace;
pub mod evaluator;
pub mod manual_transition_job;
mod metadata_boundary;
@@ -31,3 +32,8 @@ pub mod tier_free_version_recovery;
pub mod tier_last_day_stats;
pub mod tier_sweeper;
pub mod transition_transaction;
pub(crate) use durable_namespace::{
DurableIlmRecordCheckpoint, ILM_META_PREFIX, ValidatedDurableIlmRecord, classify_durable_ilm_record,
validate_durable_ilm_record,
};
crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs
+17 -1
View File
@@ -20,6 +20,7 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::TIER_DELETE_JOURNAL_NAMESPACE;
use crate::bucket::lifecycle::runtime_boundary;
use crate::bucket::lifecycle::tier_sweeper::{
Jentry, TierDeleteJournalState, TierDeleteSourceIdentity,
@@ -49,7 +50,7 @@ const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3;
const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4;
const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5;
pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/";
pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = TIER_DELETE_JOURNAL_NAMESPACE.prefix;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
@@ -432,6 +433,21 @@ async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jen
)
.await?;
}
let path = tier_delete_journal_object_name(je);
let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?;
let target_pool_indices = api
.record_durable_ilm_decommission_terminal_target_pools(&path, &data)
.await
.map_err(std::io::Error::other)?;
if let Some(target_pool_indices) = target_pool_indices {
for target_pool_idx in target_pool_indices {
match config_boundary::delete_config(api.pools[target_pool_idx].clone(), &path).await {
Ok(()) | Err(Error::ConfigNotFound) => {}
Err(err) => return Err(std::io::Error::other(err)),
}
}
return Ok(());
}
remove_tier_delete_journal_entry(api, je).await
}
crates/ecstore/src/bucket/lifecycle/transition_transaction.rs
+22 -14
View File
@@ -21,6 +21,7 @@ use tracing::{debug, warn};
use uuid::Uuid;
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE;
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::lifecycle::tier_sweeper::{
delete_confirmed_transition_candidate_exact_with_lease_idempotent,
@@ -42,7 +43,7 @@ const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(6
const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1";
pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions";
pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = "ilm/transition-transactions/records";
pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix;
pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024;
pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
@@ -584,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(
@@ -596,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),
@@ -813,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)
}
@@ -849,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),
@@ -872,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),
@@ -881,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,
@@ -907,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 => {
@@ -925,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) => {
@@ -958,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
+19
View File
@@ -406,6 +406,25 @@ where
Ok(data)
}
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
where
S: EcstoreObjectIO,
{
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
+1646 -67
View File
File diff suppressed because it is too large Load Diff
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;
@@ -4248,7 +4248,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
+1090 -3
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/store/rebalance.rs
+3 -1
View File
@@ -2301,7 +2301,9 @@ mod tests {
#[serial_test::serial]
async fn peer_pool_meta_reload_keeps_active_worker_progress_over_newer_snapshot() {
let (_temp_dir, store, shutdown) = setup_multi_pool_test_store("pool-meta-reload-worker", &[2]).await;
*store.decommission_cancelers.write().await = vec![Some(CancellationToken::new())];
*store.decommission_cancelers.write().await = vec![Some(crate::core::pools::DecommissionCanceler::new_for_test(
CancellationToken::new(),
))];
let worker_time = OffsetDateTime::now_utc();
let newer_time = worker_time + TimeDuration::seconds(30);