Compare commits

..

26 Commits

Author SHA1 Message Date
houseme d632d98a62 Merge branch 'main' into overtrue/fix-1911-ilm-metadata-decommission 2026-08-23 12:27:46 +08:00
overtrue b91cb365c4 fix(ecstore): remove redundant clone in test 2026-08-23 09:50:28 +08:00
overtrue 32436e322a fix(ecstore): clear rio-v2 lint errors 2026-08-23 05:27:57 +08:00
overtrue 12f56c9ebd fix(ecstore): fence decommission cancellation 2026-08-23 05:22:29 +08:00
overtrue acc37b49c8 test(ecstore): cover final sweep cancel fence 2026-08-23 03:21:15 +08:00
overtrue 2379bfb5a3 fix(ecstore): fence final decommission sweep 2026-08-23 03:15:33 +08:00
overtrue 4527b003cc fix(ecstore): remove duplicate decommission error helper 2026-08-23 03:00:41 +08:00
overtrue c83aa7f974 Merge remote-tracking branch 'origin/main' into overtrue/fix-1911-ilm-metadata-decommission
# Conflicts:
#	crates/ecstore/src/core/pools.rs
2026-08-23 02:58:32 +08:00
overtrue b6f135f0fe fix(ecstore): drop removed decommission test import 2026-08-22 23:28:14 +08:00
overtrue b07f0a92a8 merge: sync main into ILM decommission fix 2026-08-22 22:55:14 +08:00
overtrue 34a52a1e7d fix(ecstore): restore decommission test imports 2026-08-22 19:13:21 +08:00
overtrue fe2d166cd3 fix(ecstore): distinguish active ILM target cleanup 2026-08-22 18:43:05 +08:00
overtrue c687f95260 fix(ecstore): preserve active ILM source journals 2026-08-22 18:43:05 +08:00
overtrue fe66aa509e test(ecstore): isolate durable ILM scenario stack 2026-08-22 18:43:05 +08:00
overtrue 918686bd4e test(ecstore): compile multi-source ILM recovery 2026-08-22 18:43:05 +08:00
overtrue a9cbf82943 test(ecstore): serialize multi-source ILM recovery 2026-08-22 18:43:05 +08:00
overtrue ee7a82ec1f test(ecstore): cover durable ILM recovery boundaries 2026-08-22 18:43:05 +08:00
overtrue 3229ce05e3 fix(ecstore): repair durable ILM receipt recovery 2026-08-22 18:43:04 +08:00
overtrue 7428f3138c fix(ecstore): harden durable ILM cursor receipts 2026-08-22 18:43:04 +08:00
overtrue cf0a78ce11 fix(ecstore): avoid terminal receipt shadowing 2026-08-22 18:43:04 +08:00
overtrue 7bda7f4825 fix(ecstore): re-export durable ILM checkpoint 2026-08-22 18:43:04 +08:00
overtrue 20a8c443af fix(ecstore): anchor decommission ILM receipts 2026-08-22 18:43:04 +08:00
overtrue 948ee1fc52 fix(ecstore): close ILM receipt recovery gaps 2026-08-22 18:43:04 +08:00
overtrue ecd779b6f3 fix(ecstore): track ILM recovery across decommission 2026-08-22 18:43:04 +08:00
overtrue 619fae2512 fix(ecstore): verify ILM metadata before decommission 2026-08-22 18:43:04 +08:00
overtrue 4f9d22bda7 fix(ecstore): migrate ILM metadata during decommission 2026-08-22 18:43:04 +08:00
22 changed files with 4118 additions and 1505 deletions
+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]]
+6 -3
View File
@@ -39,10 +39,11 @@ jobs:
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -88,10 +89,11 @@ jobs:
# either casing.
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -176,10 +178,11 @@ jobs:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -9136,6 +9136,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(1));
assert!(loaded.lease_expires_at_unix_nanos > 0);
let token = loaded
.report
@@ -9154,6 +9155,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(2));
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 +9235,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(3));
let token = checkpointed
.report
.continuation_token
File diff suppressed because it is too large Load Diff
@@ -24,6 +24,10 @@ 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 +38,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 +199,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_if = "Option::is_none")]
pub cursor_revision: Option<u64>,
pub report: ManualTransitionRunReport,
pub queue_snapshot: ManualTransitionQueueSnapshot,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -224,6 +230,7 @@ impl ManualTransitionJobRecord {
created_at_unix_nanos: now,
updated_at_unix_nanos: now,
completed_at_unix_nanos: None,
cursor_revision: Some(0),
report: ManualTransitionRunReport {
bucket: bucket.to_string(),
prefix: options.prefix.clone(),
@@ -238,7 +245,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 +323,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 +366,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 +387,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 +405,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 +445,18 @@ 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) {
if self.report.continuation_token != report.continuation_token {
self.cursor_revision = Some(self.cursor_revision.unwrap_or(0).saturating_add(1));
}
self.report.merge_scan_report_preserving_worker(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) {
@@ -1109,7 +1127,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 +1161,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 +1173,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 +1612,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 +1624,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 +1663,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 +1681,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 +1975,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),
@@ -2577,6 +2601,25 @@ mod tests {
assert!(decoded.report.tier_failure_by_reason.is_empty());
}
#[test]
fn manual_transition_job_record_decodes_legacy_cursor_without_revision() {
let options = ManualTransitionRunOptions::default();
let record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
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, None);
}
#[test]
fn manual_transition_job_record_rejects_unknown_report_fields() {
let options = ManualTransitionRunOptions::default();
@@ -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,
};
@@ -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
}
@@ -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)
}
+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`.
File diff suppressed because it is too large Load Diff
-13
View File
@@ -1950,19 +1950,6 @@ mod tests {
assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &allowed_missing));
}
#[test]
fn test_decommission_cleanup_preflight_accepts_migrated_free_version_consumed_from_source() {
let migrated = cleanup_test_file_info("object.txt", Uuid::from_u128(1), "migrated");
let mut free_version = cleanup_test_file_info("object.txt", Uuid::from_u128(2), "tier-cleanup");
free_version.deleted = true;
free_version.set_tier_free_version();
let expected = cleanup_test_versions(vec![migrated.clone(), free_version.clone()]);
let current = cleanup_test_versions(vec![migrated]);
let allowed_missing = vec![source_cleanup_version_identity(&free_version)];
assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &allowed_missing));
}
#[test]
fn test_decommission_cleanup_preflight_rejects_unexpected_missing_version() {
let migrated = cleanup_test_file_info("object.txt", Uuid::from_u128(1), "migrated");
+6 -38
View File
@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
///
/// # Returns
/// A Result containing the BitrotWriterWrapper or an error
/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
///
/// A known length is grown by one checksum per shard so the on-disk file size
/// matches what the bitrot writer emits. A negative length is the
/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
/// a fixed body length when locating the authenticated trailer. Clamping it
/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
/// because a genuinely empty object still means an empty body.
fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
if length <= 0 {
return length;
}
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
}
pub async fn create_bitrot_writer(
is_inline_buffer: bool,
disk: Option<&DiskStore>,
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
let writer = if is_inline_buffer {
CustomWriter::new_inline_buffer()
} else if let Some(disk) = disk {
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
let length = if length > 0 {
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
} else {
0
};
let file = disk.create_file("", volume, path, length).await?;
#[cfg(feature = "hotpath")]
@@ -833,25 +820,6 @@ mod tests {
use rustfs_rio::ChunkReader;
use std::collections::VecDeque;
#[test]
fn bitrot_create_file_size_grows_known_length_by_checksums() {
// 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
}
#[test]
fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
// SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
// put_file_stream receiver relies on a non-positive size to parse the auth
// trailer from the stream tail, so the sentinel must survive untouched.
assert_eq!(
bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
);
}
struct TestChunkReader {
chunks: VecDeque<Bytes>,
}
-297
View File
@@ -227,70 +227,6 @@ pub(super) fn restore_commit_operation_id_from_metadata(metadata: &HashMap<Strin
restore_operation_id_from_metadata(metadata)
}
async fn inspect_decommission_tier_free_version_target(
disk: &DiskStore,
bucket: &str,
object: &str,
source: &FileInfo,
) -> Result<bool> {
let raw = match disk.read_xl(bucket, object, false).await {
Ok(raw) => raw,
Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => return Ok(false),
Err(err) => return Err(err.into()),
};
let meta = FileMeta::load(&raw.buf)?;
let source_version_id = source.version_id.filter(|version_id| !version_id.is_nil());
let mut matching_count = 0;
let mut all_matching_versions_equivalent = true;
for existing in meta
.versions
.iter()
.filter(|version| version.header.version_id.filter(|version_id| !version_id.is_nil()) == source_version_id)
{
matching_count += 1;
let existing = existing.into_fileinfo(bucket, object, true)?;
existing.validate_for_metadata_read()?;
if !existing.tier_free_version() || !crate::store::tiered_data_movement_source_matches(source, &existing)? {
all_matching_versions_equivalent = false;
}
}
if matching_count == 0 {
return Ok(false);
}
if matching_count == 1 && all_matching_versions_equivalent {
return Ok(true);
}
Err(StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
source_version_id.map(|version_id| version_id.to_string()).unwrap_or_default(),
)
.into())
}
fn ensure_decommission_tier_free_version_commit_fence(bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {
if opts
.namespace_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
|| opts
.bucket_lifecycle_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
{
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "decommission_tier_free_version_commit",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
Ok(())
}
impl SetDisks {
pub(super) async fn require_current_restore_operation_id(
&self,
@@ -4729,98 +4665,6 @@ fn resolve_delete_version_state(opts: &ObjectOptions, goi: &ObjectInfo, version_
}
impl SetDisks {
/// Publish an internal tier free-version record without changing its
/// delete-marker shape or remote-tier identity. The caller holds the
/// source and target object locks; a write quorum is required before the
/// source cleanup may remove the original record.
#[tracing::instrument(skip(self, fi, opts))]
pub(crate) async fn decommission_tier_free_version(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
opts: &ObjectOptions,
) -> Result<()> {
if !fi.deleted || !fi.tier_free_version() {
return Err(Error::other("decommission tier free-version write requires a free version record"));
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
self.validate_decommission_tier_free_version_target(bucket, object, fi)
.await?;
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
let disks = self.disks.read().await.clone();
let write_quorum = self.default_write_quorum();
let futures = disks.into_iter().map(|disk| {
let file_info = fi.clone();
async move {
if let Some(disk) = disk {
disk.write_metadata("", bucket, object, file_info).await
} else {
Err(DiskError::DiskNotFound)
}
}
});
let mut errs = Vec::new();
for result in join_all(futures).await {
match result {
Ok(_) => errs.push(None),
Err(err) => errs.push(Some(err)),
}
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object)
}
pub(crate) async fn validate_decommission_tier_free_version_target(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
) -> Result<()> {
// The caller holds the source and target object locks. Inspect every
// target disk before an idempotent return or metadata fan-out so a
// sub-quorum conflict cannot be hidden by a successful quorum.
let disks = self.disks.read().await.clone();
let preflight = disks
.iter()
.flatten()
.map(|disk| inspect_decommission_tier_free_version_target(disk, bucket, object, fi));
for result in join_all(preflight).await {
result?;
}
Ok(())
}
pub(crate) async fn has_decommission_tier_free_version_write_quorum(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
opts: &ObjectOptions,
) -> Result<bool> {
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
let disks = self.disks.read().await.clone();
let preflight = disks.iter().map(|disk| async {
match disk {
Some(disk) => inspect_decommission_tier_free_version_target(disk, bucket, object, fi).await,
None => Ok(false),
}
});
let mut equivalent = 0;
for result in join_all(preflight).await {
if result? {
equivalent += 1;
}
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
Ok(equivalent >= self.default_write_quorum())
}
#[tracing::instrument(skip(self, fi, opts))]
pub(crate) async fn decommission_tiered_object(
&self,
@@ -10043,147 +9887,6 @@ mod tests {
assert_ne!(updated.erasure.distribution, original.erasure.distribution);
}
#[tokio::test]
async fn decommission_tier_free_version_preserves_remote_identity() {
let set_disks = make_local_bucket_test_set_disks().await;
let bucket = "free-version-decommission";
let object = "object.txt";
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("target bucket should exist before free-version migration");
let version_id = Uuid::new_v4();
let mut free_version = FileInfo {
name: object.to_string(),
volume: bucket.to_string(),
version_id: Some(version_id),
mod_time: Some(time::OffsetDateTime::now_utc()),
deleted: true,
transition_tier: "WARM-TIER".to_string(),
transitioned_objname: "remote/object".to_string(),
..Default::default()
};
free_version.set_tier_free_version();
// Decoded free versions always carry the on-disk free-version
// suffix alongside the in-memory tier marker; mirror that here so
// the record satisfies delete-marker metadata validation.
rustfs_utils::http::metadata_compat::insert_str(
&mut free_version.metadata,
rustfs_utils::http::metadata_compat::SUFFIX_FREE_VERSION,
String::new(),
);
set_disks
.decommission_tier_free_version(bucket, object, &free_version, &ObjectOptions::default())
.await
.expect("free-version metadata should reach the target quorum");
set_disks
.decommission_tier_free_version(bucket, object, &free_version, &ObjectOptions::default())
.await
.expect("replaying the same free-version metadata should be idempotent");
let versions = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("migrated free-version metadata should decode")
.expect("migrated free-version metadata should exist");
let migrated = versions
.versions
.iter()
.find(|version| version.version_id == Some(version_id))
.expect("free version should be present on the target");
assert_eq!(
versions
.versions
.iter()
.filter(|version| version.version_id == Some(version_id))
.count(),
1
);
assert!(migrated.tier_free_version());
assert_eq!(migrated.transition_tier, "WARM-TIER");
assert_eq!(migrated.transitioned_objname, "remote/object");
}
#[tokio::test]
async fn decommission_tier_free_version_resume_requires_write_quorum() {
let set_disks = make_local_bucket_test_set_disks_with_drive_count(4).await;
let bucket = "free-version-decommission-resume";
let object = "object.txt";
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("target bucket should exist before free-version migration");
let mut free_version = FileInfo {
name: object.to_string(),
volume: bucket.to_string(),
version_id: Some(Uuid::new_v4()),
mod_time: Some(time::OffsetDateTime::now_utc()),
deleted: true,
transition_tier: "WARM-TIER".to_string(),
transitioned_objname: "remote/object".to_string(),
..Default::default()
};
free_version.set_tier_free_version();
// Decoded free versions always carry the on-disk free-version
// suffix alongside the in-memory tier marker; mirror that here so
// the record satisfies delete-marker metadata validation.
rustfs_utils::http::metadata_compat::insert_str(
&mut free_version.metadata,
rustfs_utils::http::metadata_compat::SUFFIX_FREE_VERSION,
String::new(),
);
let opts = ObjectOptions::default();
let disks = set_disks.get_disks_internal().await;
for disk in disks.iter().take(2).flatten() {
disk.write_metadata("", bucket, object, free_version.clone())
.await
.expect("partial first attempt should leave equivalent metadata");
}
assert!(
!set_disks
.has_decommission_tier_free_version_write_quorum(bucket, object, &free_version, &opts)
.await
.expect("partial target metadata should remain valid"),
"write-quorum-minus-one must not be accepted as an idempotent migration"
);
disks[2]
.as_ref()
.expect("third target disk should be online")
.write_metadata("", bucket, object, free_version.clone())
.await
.expect("third equivalent target write should complete quorum");
assert!(
set_disks
.has_decommission_tier_free_version_write_quorum(bucket, object, &free_version, &opts)
.await
.expect("write-quorum target metadata should remain valid")
);
}
#[test]
fn decommission_tier_free_version_commit_rejects_lost_fence() {
let opts = ObjectOptions {
namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()),
..Default::default()
};
let err = ensure_decommission_tier_free_version_commit_fence("bucket", "object", &opts)
.expect_err("lost target lock must fail the free-version commit");
assert!(matches!(
err,
Error::NamespaceLockQuorumUnavailable {
mode: "decommission_tier_free_version_commit",
required: 1,
achieved: 0,
..
}
));
}
#[test]
fn test_resolve_tiered_decommission_write_quorum_result_allows_successful_quorum() {
let errs = vec![None, None, Some(DiskError::DiskNotFound), None];
+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(())
}
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -151,7 +151,7 @@ pub(crate) mod init_format;
pub(crate) mod list_objects;
mod multipart;
mod object;
pub(crate) use object::{ObjectLockDiagGuard, SourceCleanupMutationFence, tiered_data_movement_source_matches};
pub(crate) use object::{ObjectLockDiagGuard, SourceCleanupMutationFence};
pub use object::{
PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError,
SnapshotConsistencyError,
+15 -143
View File
@@ -490,82 +490,6 @@ fn decommission_mutation_fence_for_test(
.map(|hook| hook.fence.clone())
}
#[cfg(test)]
struct DecommissionFreeVersionSourceRaceState {
bucket: String,
object: String,
arrived: tokio::sync::Notify,
release: tokio::sync::Notify,
}
#[cfg(test)]
pub(crate) struct DecommissionFreeVersionSourceRaceBarrier {
state: Arc<DecommissionFreeVersionSourceRaceState>,
}
#[cfg(test)]
static DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER: std::sync::OnceLock<
std::sync::Mutex<Option<Arc<DecommissionFreeVersionSourceRaceState>>>,
> = std::sync::OnceLock::new();
#[cfg(test)]
impl DecommissionFreeVersionSourceRaceBarrier {
pub(crate) fn install(bucket: &str, object: &str) -> Self {
let state = Arc::new(DecommissionFreeVersionSourceRaceState {
bucket: bucket.to_string(),
object: object.to_string(),
arrived: tokio::sync::Notify::new(),
release: tokio::sync::Notify::new(),
});
let mut slot = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison");
assert!(slot.is_none(), "decommission free-version source race barrier must be unique");
*slot = Some(Arc::clone(&state));
Self { state }
}
pub(crate) async fn wait_until_paused(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
.expect("decommission should pause before acquiring the free-version source lock");
}
pub(crate) fn release(&self) {
self.state.release.notify_one();
}
}
#[cfg(test)]
impl Drop for DecommissionFreeVersionSourceRaceBarrier {
fn drop(&mut self) {
self.state.release.notify_one();
let mut slot = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison");
if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
*slot = None;
}
}
}
#[cfg(test)]
async fn pause_decommission_free_version_before_source_lock(bucket: &str, object: &str) {
let state = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison")
.as_ref()
.filter(|state| state.bucket == bucket && state.object == object)
.cloned();
if let Some(state) = state {
state.arrived.notify_one();
state.release.notified().await;
}
}
pub(crate) struct SourceCleanupMutationFence {
guard: ObjectLockDiagGuard,
source_lock_covered: bool,
@@ -1570,15 +1494,13 @@ fn is_equivalent_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo,
&& source_actual_size == target_actual_size
}
pub(crate) fn tiered_data_movement_source_matches(
fn tiered_data_movement_source_matches(
expected: &rustfs_filemeta::FileInfo,
current: &rustfs_filemeta::FileInfo,
) -> Result<bool> {
let expected_backend = crate::services::tier::tier::tier_destination_id_from_metadata(&expected.metadata)?;
let current_backend = crate::services::tier::tier::tier_destination_id_from_metadata(&current.metadata)?;
Ok(expected.version_id == current.version_id
&& expected.deleted == current.deleted
&& expected.tier_free_version() == current.tier_free_version()
&& expected.data_dir == current.data_dir
&& expected.mod_time == current.mod_time
&& expected.size == current.size
@@ -1592,15 +1514,6 @@ pub(crate) fn tiered_data_movement_source_matches(
&& expected_backend == current_backend)
}
fn decommission_free_version_overwrite_error(bucket: &str, object: &str, version_id: Option<Uuid>) -> Error {
StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
version_id.map(|id| id.to_string()).unwrap_or_default(),
)
.into()
}
fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool {
target_pool_idx != src_pool_idx
}
@@ -2308,23 +2221,6 @@ impl ECStore {
)
}
async fn has_equivalent_data_movement_tier_free_version(
&self,
bucket: &str,
object: &str,
source: &rustfs_filemeta::FileInfo,
opts: &ObjectOptions,
target_pool_idx: usize,
) -> Result<bool> {
let pool = self
.pools
.get(target_pool_idx)
.ok_or_else(|| Error::other(format!("invalid tiered data movement target pool {target_pool_idx}")))?;
pool.get_disks_by_key(object)
.has_decommission_tier_free_version_write_quorum(bucket, object, source, opts)
.await
}
fn resolve_decommission_target_pool_idx_result(result: Result<usize>, bucket: &str, object: &str) -> Result<usize> {
result.map_err(|err| Error::other(format!("failed to select decommission target pool for {bucket}/{object}: {err}")))
}
@@ -2344,10 +2240,6 @@ impl ECStore {
check_put_object_args(bucket, object)?;
let mut opts = opts.clone();
let is_free_version = fi.tier_free_version();
if is_free_version {
opts.incl_free_versions = true;
}
let bucket_incarnation_fence = if is_meta_bucketname(bucket) {
None
} else {
@@ -2385,10 +2277,6 @@ impl ECStore {
&object,
)?
};
#[cfg(test)]
if is_free_version {
pause_decommission_free_version_before_source_lock(bucket, logical_object).await;
}
let _object_guards = self
.acquire_data_movement_object_write_locks(bucket, &object, opts.src_pool_idx, idx, &mut opts)
.await?;
@@ -2406,7 +2294,7 @@ impl ECStore {
versions
.versions
.iter()
.find(|current| current.version_id == fi.version_id && current.tier_free_version() == is_free_version)
.find(|current| current.version_id == fi.version_id && !current.tier_free_version())
})
.ok_or_else(|| to_object_err(StorageError::FileNotFound, vec![bucket, object.as_str()]))?;
if !tiered_data_movement_source_matches(fi, current_source)? {
@@ -2421,40 +2309,24 @@ impl ECStore {
.get_available_pool_idx_excluding(bucket, &object, fi.size, opts.src_pool_idx)
.await;
let target_pool_idx = resolve_data_movement_resume_target_pool(idx, resume_target_pool_idx, opts.src_pool_idx);
if is_free_version && target_pool_idx == opts.src_pool_idx {
return Err(Error::DiskFull);
}
let equivalent = if is_free_version {
self.has_equivalent_data_movement_tier_free_version(bucket, &object, &fi, &opts, target_pool_idx)
.await?
} else {
self.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, target_pool_idx)
.await?
};
if equivalent {
return Ok(());
}
return Err(decommission_free_version_overwrite_error(bucket, &object, fi.version_id));
}
let result = if is_free_version {
if self
.has_equivalent_data_movement_tier_free_version(bucket, &object, &fi, &opts, idx)
.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, target_pool_idx)
.await?
{
return Ok(());
}
self.pools[idx]
.get_disks_by_key(&object)
.decommission_tier_free_version(bucket, &object, &fi, &opts)
.await
} else {
self.pools[idx]
.get_disks_by_key(&object)
.decommission_tiered_object(bucket, &object, &fi, &opts)
.await
};
return Err(StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
opts.version_id.clone().unwrap_or_default(),
));
}
let result = self.pools[idx]
.get_disks_by_key(&object)
.decommission_tiered_object(bucket, &object, &fi, &opts)
.await;
if matches!(result, Err(Error::PreconditionFailed)) {
if self
.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, idx)
-19
View File
@@ -90,21 +90,6 @@ fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
pub const TRANSITION_COMPLETE: &str = "complete";
pub const TRANSITION_PENDING: &str = "pending";
/// xl.meta key marking a tier free-version record.
///
/// A free version is a delete-marker-shaped cleanup hint appended by
/// [`MetaObject::delete_version`] when a version whose remote transition
/// completed is removed from xl.meta; it carries the remote tier identity for
/// an idempotent remote delete and is never a user-visible version
/// (`num_versions` excludes it). While the record exists it is consumed by the
/// lifecycle free-version recovery scan and the usage scanner, which re-enqueue
/// the pending remote delete, and by heal metadata walks. On S3 and lifecycle
/// delete paths the same obligation is also carried by a committed tier-journal
/// entry; deletes without such an entry (for example a removed version whose
/// transition state decodes as unknown) rely on this record alone until the
/// worker removes it after a successful remote delete. Decommission preserves
/// the record and its remote identity on the target pool before source cleanup
/// — see docs/architecture/decommission-compatibility.md.
pub const FREE_VERSION: &str = "free-version";
pub const TRANSITION_STATUS: &str = "transition-status";
@@ -462,10 +447,6 @@ impl FileMeta {
};
if let Some(fidx) = existing_idx {
let existing = self.versions[fidx].parse_version_meta()?;
if existing.free_version() != version.free_version() {
return Err(Error::other("cannot replace a free version with a non-free version"));
}
return self.set_idx(fidx, version);
}
-9
View File
@@ -2725,15 +2725,6 @@ impl MetaObject {
self.meta_sys.retain(|k, _| !k.starts_with("X-Amz-Restore"));
}
/// Builds the free-version cleanup record appended when a transitioned
/// version is removed from xl.meta. The record keeps the remote tier
/// identity so the lifecycle worker can issue the idempotent remote delete
/// and only then remove the record; until then the recovery scan and the
/// usage scanner keep re-enqueueing it. S3 and lifecycle deletes also
/// persist a committed tier-journal entry for the same remote delete. The
/// decommission path copies this record unchanged before source cleanup,
/// including when the transition state is unknown — see
/// docs/architecture/decommission-compatibility.md.
pub fn init_free_version(&self, fi: &FileInfo) -> Result<(FileMetaVersion, bool)> {
if fi.skip_tier_free_version() {
return Ok((FileMetaVersion::default(), false));
+25 -86
View File
@@ -16,14 +16,14 @@
//!
//! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
//! kills the active node while this test continuously decrypts through a
//! surviving standby. KV2 and Transit must recover after the bounded circuit
//! interval, use a bounded number of attempts, and leave the circuit and
//! in-flight gauges at zero after a new leader is elected.
//! surviving standby. KV2 and Transit requests must remain successful, use a
//! bounded number of attempts, and leave the circuit and in-flight gauges at
//! zero after a new leader is elected.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use metrics_util::MetricKind;
@@ -43,11 +43,6 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
const MAX_ATTEMPTS: u32 = 10;
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
// The circuit remains open for 30s after five failed attempts.
const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
type MetricEntry = (
metrics_util::CompositeKey,
@@ -69,7 +64,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
backend,
backend_config,
allow_insecure_dev_defaults: true,
timeout: ATTEMPT_TIMEOUT,
timeout: Duration::from_secs(2),
retry_attempts: MAX_ATTEMPTS,
enable_cache: false,
..KmsConfig::default()
@@ -169,31 +164,14 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
.sum()
}
async fn wait_for_count(
counter: &AtomicU64,
failure: &Mutex<Option<String>>,
minimum: u64,
description: &str,
timeout: Duration,
) {
tokio::time::timeout(timeout, async {
async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
tokio::time::timeout(Duration::from_secs(20), async {
while counter.load(Ordering::SeqCst) < minimum {
if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
panic!(
"{description} worker failed after {} successful decrypts: {error}",
counter.load(Ordering::SeqCst)
);
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap_or_else(|_| {
panic!(
"timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
counter.load(Ordering::SeqCst)
)
});
.unwrap_or_else(|_| panic!("timed out waiting for {description}"));
}
async fn wait_for_file(path: &Path, description: &str) {
@@ -211,8 +189,7 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
request: DecryptRequest,
expected: Vec<u8>,
completed: Arc<AtomicU64>,
allow_failover_errors: Arc<AtomicBool>,
failure: Arc<Mutex<Option<String>>>,
failed: Arc<AtomicBool>,
stop: CancellationToken,
) {
while !stop.is_cancelled() {
@@ -220,18 +197,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
Ok(response) if response.plaintext == expected => {
completed.fetch_add(1, Ordering::SeqCst);
}
Ok(_) => {
*failure.lock().expect("decrypt failure lock poisoned") =
Some("decrypt returned unexpected plaintext".to_string());
return;
}
Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
if allow_failover_errors.load(Ordering::SeqCst) =>
{
tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
}
Err(error) => {
*failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
Ok(_) | Err(_) => {
failed.store(true, Ordering::SeqCst);
return;
}
}
@@ -329,9 +296,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
);
let stop = CancellationToken::new();
let allow_failover_errors = Arc::new(AtomicBool::new(false));
let kv2_failure = Arc::new(Mutex::new(None));
let transit_failure = Arc::new(Mutex::new(None));
let failed = Arc::new(AtomicBool::new(false));
let kv2_completed = Arc::new(AtomicU64::new(0));
let transit_completed = Arc::new(AtomicU64::new(0));
let kv2_worker = tokio::spawn(decrypt_loop(
@@ -339,8 +304,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
kv2_request,
kv2_data_key.plaintext_key,
Arc::clone(&kv2_completed),
Arc::clone(&allow_failover_errors),
Arc::clone(&kv2_failure),
Arc::clone(&failed),
stop.clone(),
));
let transit_worker = tokio::spawn(decrypt_loop(
@@ -348,21 +312,12 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
transit_request,
transit_data_key.plaintext_key,
Arc::clone(&transit_completed),
Arc::clone(&allow_failover_errors),
Arc::clone(&transit_failure),
Arc::clone(&failed),
stop.clone(),
));
wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
wait_for_count(
&transit_completed,
&transit_failure,
2,
"two healthy Transit decrypts",
HEALTHY_PROGRESS_TIMEOUT,
)
.await;
allow_failover_errors.store(true, Ordering::SeqCst);
wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
wait_for_file(&elected, "the replacement Vault leader").await;
@@ -371,39 +326,18 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
wait_for_count(
&kv2_completed,
&kv2_failure,
kv2_after_election,
"post-failover KV2 decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
wait_for_count(
&transit_completed,
&transit_failure,
transit_after_election,
"post-failover Transit decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
stop.cancel();
kv2_worker.await.expect("KV2 decrypt worker must join");
transit_worker.await.expect("Transit decrypt worker must join");
assert!(
kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
"no KV2 decrypt may fail or return different plaintext"
);
assert!(
transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
"no Transit decrypt may fail or return different plaintext"
);
assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
}
#[test]
#[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
@@ -415,6 +349,11 @@ fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
});
let snapshot = snapshotter.snapshot().into_vec();
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
0,
"a bounded leader election must not open the circuit"
);
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
0,
@@ -153,99 +153,6 @@ No migration step is required for these decisions because this note documents th
current RustFS behavior. Changing either decision later requires an operator
compatibility note and updated characterization tests.
## Tier Free Versions During Decommission
A tier free version is an internal xl.meta record (`rustfs_filemeta::FREE_VERSION`,
flagged `XL_FLAG_FREE_VERSION`) shaped like a delete marker. It is created by
`MetaObject::init_free_version` when a version whose remote transition completed is
deleted locally: the visible version is removed and the record keeps the remote-tier
identity (tier, object name, version id, state, destination id) needed for an
idempotent remote delete. Free versions are not user-visible versions; `num_versions`
and all listing/GET paths exclude them.
### Lifecycle And Consumers
Creation: any local delete that removes a version whose transition status is
`complete` appends the record via `MetaObject::delete_version`
`init_free_version` (skipped only when `skip_tier_free_version` is set, as on
data-movement copies). The same deletes also persist a durable tier-journal
entry on every user-facing path: S3 single deletes (`execute_delete_object`
`delete_object_with_tier_delete_journal`), S3 batch deletes, lifecycle expiry,
and lifecycle delete-all all prepare and commit a journal entry around the
delete. A journal entry is omitted when the removed version's transition state
decodes as `TransitionVersionState::Unknown`, or on internal journal-less
delete paths that never touch transitioned user objects.
Consumption while the record exists: the background recovery loop started by
`init_background_expiry` (spawned by `spawn_tier_free_version_recovery_once`,
enabled by default) scans disks for pending records and re-enqueues them; the
usage scanner does the same; the lifecycle worker then deletes the remote tier
object idempotently and only afterwards removes the local record. Heal walks
include free-version records in metadata healing. Transition planning,
replication, restore, GET, listings, and usage aggregation never depend on
them.
### Decommission Handling
The exact decommission inventory loader (`load_file_info_versions_exact` via
`get_all_file_info_versions`) keeps free-version records inline in `versions`.
The migration loop handles them before lifecycle expiry and delete-marker
shortcuts. It selects a target pool using the free-version-aware lookup, then
writes the original free record to every target disk with the normal metadata
write quorum. The free-version marker, local version id, transition identity,
transition state, and destination id are preserved at the FileInfo/metadata
boundary.
The source record is physically removed only after the target write quorum has
committed and the source cleanup preflight still matches the exact inventory.
If the lifecycle worker has already completed the remote delete and removed the
source record before decommission acquires the source lock, decommission records
that identity as already consumed and treats the missing source record as safe.
If target capacity, metadata validation, lock fencing, or quorum fails, the
source record remains and the entry records `state = "free_version_retained"`
with reason `tier_free_version_migration_failed`; the worker retries the
operation on a later pass. A target record with the same version id is accepted
only when its free-version identity matches; a conflicting ordinary version or
different free record is an overwrite error. This makes retries idempotent and
prevents a free record from replacing a user-visible version.
### Reference-Audit Result
After migration, user-facing GET/list/transition/replication/restore paths still
exclude the record. Recovery, usage scanning, lifecycle tier cleanup, and heal
continue to see it when they request free versions, so an unresolved remote
delete remains actionable on the target pool. The committed tier journal remains
an independent retry source where one exists; it is not used as a reason to drop
the xl.meta record. In particular, `Unknown` transition state records are
migrated unchanged rather than discarded: the lifecycle worker retains them if
remote identity validation cannot make a delete request.
Each migrated record emits `state = "free_version_migrated"` with reason
`tier_free_version_migrated`. A record consumed before migration emits
`state = "free_version_consumed"` with reason
`tier_free_version_already_consumed`. Each failed record emits the retained state
and failure reason above. The entry also emits a disposition summary with
migrated, consumed, retained, and total counts. The final decommission sweep uses
the exact loader, counts free records still present, and emits one retained
record/reason for each unresolved free version before failing the sweep. This
makes successful migration, completed cleanup, and retained cleanup obligations
visible instead of silently omitting free records.
No new S3-visible version or admin response field is needed: free versions remain
internal and are never counted as user-visible versions. The structured
`decommission_entry` events are the operational status surface for the
free-version disposition; the existing decommission item/failed counters still
report the enclosing object migration result.
Regression guard:
- `decommission_tier_free_version_preserves_remote_identity`
- `decommission_tier_free_version_resume_requires_write_quorum`
- `decommission_tier_free_version_commit_rejects_lost_fence`
- `test_decommission_cleanup_preflight_accepts_migrated_free_version_consumed_from_source`
- `decommission_entry_skips_cleanup_only_marker_when_free_version_is_present`
- `decommission_entry_rejects_subquorum_free_version_conflict_and_retains_source`
## Regression Guard
The queued multi-pool contract is guarded by:
+1 -1
View File
@@ -241,7 +241,7 @@ env \
RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
cargo test -p rustfs-kms --test vault_ha_failover_live \
vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
--ignored --nocapture --test-threads=1 &
TEST_PID=$!