diff --git a/.config/test-count-baseline.txt b/.config/test-count-baseline.txt index 016adc70d..0e61ce116 100644 --- a/.config/test-count-baseline.txt +++ b/.config/test-count-baseline.txt @@ -20,7 +20,7 @@ rustfs-connect-symbol-catalog 6 0 rustfs-credentials 24 0 rustfs-crypto 67 0 rustfs-data-usage 54 0 -rustfs-ecstore 6581 1 +rustfs-ecstore 6585 1 rustfs-extension-schema 23 0 rustfs-filemeta 306 0 rustfs-heal 856 3 diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index da079d72a..e6822bd34 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -10331,6 +10331,42 @@ mod tests { assert_eq!(observed.committed_config_etag.as_deref(), Some("etag-new")); } + #[tokio::test] + async fn intent_advance_retries_cas_conflict_reload_lock_timeout() { + let store = Arc::new(CasConfigStore::default()); + let mutation_id = uuid::Uuid::from_u128(39); + let prepared = prepared_remove_intent("COLD-A", mutation_id); + crate::services::tier::tier_mutation_intent::save_tier_mutation_intent_record(store.clone(), &prepared) + .await + .expect("prepared intent fixture should persist"); + + let committed = committed_remove_intent("COLD-A", mutation_id, "etag-new"); + let committed_bytes = committed.encode().expect("committed intent fixture should encode"); + let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id) + .expect("record object should build"); + let winning_etag = format!("etag-{}", store.next_etag.load(Ordering::SeqCst) + 1); + store.rewrite_on_next_if_match(object.clone(), committed_bytes.clone()).await; + *store.read_timeout_after_if_match_race.lock().await = Some(object.clone()); + + let (observed, applied) = crate::services::tier::tier_mutation_intent::advance_tier_mutation_intent_record_idempotent( + store.clone(), + mutation_id, + TierMutationIntentState::Committed, + Some("etag-new".to_string()), + ) + .await + .expect("CAS conflict reload should retry its typed lock timeout"); + + assert!(!applied); + assert_eq!(observed, committed); + assert!(store.read_timeout_after_if_match_race.lock().await.is_none()); + assert_eq!( + store.objects.lock().await.get(&object).cloned(), + Some((committed_bytes, winning_etag)), + "retry must preserve the complete winning record and its object ETag" + ); + } + #[tokio::test] async fn intent_advance_reconciles_matching_commit_after_final_cas_race() { let store = Arc::new(CasConfigStore::default()); @@ -14802,6 +14838,7 @@ mod tests { objects: tokio::sync::Mutex, String)>>, legacy_state: tokio::sync::Mutex>>, if_match_race_rewrite: tokio::sync::Mutex)>>, + read_timeout_after_if_match_race: tokio::sync::Mutex>, next_etag: AtomicUsize, content_derived_etags: AtomicBool, fail_put: AtomicBool, @@ -14827,6 +14864,7 @@ mod tests { objects: tokio::sync::Mutex::new(HashMap::new()), legacy_state: tokio::sync::Mutex::new(None), if_match_race_rewrite: tokio::sync::Mutex::new(std::collections::VecDeque::new()), + read_timeout_after_if_match_race: tokio::sync::Mutex::new(None), next_etag: AtomicUsize::new(0), content_derived_etags: AtomicBool::new(false), fail_put: AtomicBool::new(false), @@ -14984,6 +15022,19 @@ mod tests { crate::api::diagnostic::current_read_version_entry_source(), rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_name(), )); + { + // The queued rewrite keeps the initial read successful. Only + // the first target read after that CAS race consumes the fault. + let race_completed = self.if_match_race_rewrite.lock().await.is_empty(); + let mut timeout = self.read_timeout_after_if_match_race.lock().await; + if race_completed && timeout.as_deref() == Some(object) { + timeout.take(); + return Err(Error::Lock(rustfs_lock::LockError::Timeout { + resource: format!("{bucket}/{object}@latest"), + timeout: std::time::Duration::from_secs(5), + })); + } + } let legacy_config_path = tier_config_path(TIER_CONFIG_LEGACY_FILE); let (data, etag) = if object == legacy_config_path { let state = self.legacy_state.lock().await; diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index eef2dcf7a..453512fb1 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -700,9 +700,25 @@ async fn advance_tier_mutation_intent_record_idempotent_at_prefix( where S: EcstoreObjectIO, { + let mut original_intent: Option = None; for attempt in 0..TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS { let (mut intent, current_etag) = - load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await?; + match load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await { + Ok(current) => current, + Err(Error::Lock(rustfs_lock::LockError::Timeout { .. })) + if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS => + { + continue; + } + Err(err) => return Err(err), + }; + if let Some(original) = &original_intent { + if !intent.same_identity_as(original) { + return Err(Error::PreconditionFailed); + } + } else { + original_intent = Some(intent.clone()); + } let advanced = intent .advance_idempotent(next, committed_config_etag.clone()) .map_err(tier_mutation_intent_store_error)?; @@ -713,7 +729,15 @@ where Ok(()) => return Ok((intent, true)), Err(Error::PreconditionFailed) => { let (mut current, _) = - load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await?; + match load_tier_mutation_intent_record_with_etag_at_prefix(api.clone(), prefix, mutation_id).await { + Ok(current) => current, + Err(Error::Lock(rustfs_lock::LockError::Timeout { .. })) + if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS => + { + continue; + } + Err(err) => return Err(err), + }; if !current.same_identity_as(&intent) { return Err(Error::PreconditionFailed); } @@ -728,6 +752,11 @@ where } return Ok((current, false)); } + Err(Error::Lock(rustfs_lock::LockError::Timeout { .. })) + if attempt + 1 < TIER_MUTATION_INTENT_ADVANCE_CAS_ATTEMPTS => + { + continue; + } Err(err) => return Err(err), } } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 4a8915434..43117f74f 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -23323,6 +23323,418 @@ mod tests { shutdown.cancel(); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_mutation_peer_commit_reloads_after_save_and_load_lock_timeouts() { + use crate::storage_api_contracts::namespace::NamespaceLocking as _; + use std::sync::atomic::Ordering; + + let temp_dir = tempfile::tempdir().expect("create contended tier commit store dir"); + let (ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-commit-contention", &[4])).await; + register_mock_tier(&store.tier_config_mgr(), "COLD-A").await; + let (candidate_digest, config_etag) = { + let tier_config_mgr = store.tier_config_mgr(); + let manager = tier_config_mgr.read().await; + let candidate_digest = tier_config_candidate_digest(&manager).expect("peer commit candidate digest should build"); + manager + .save_tiering_config(store.clone()) + .await + .expect("peer commit config fixture should persist"); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("peer commit config fixture should load"); + (candidate_digest, config_info.etag.expect("peer commit config should carry an ETag")) + }; + let mutation_id = uuid::Uuid::new_v4(); + let mut intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", candidate_digest); + intent.old_config_etag = Some(config_etag.clone()); + let prepared = handle_tier_mutation_peer_request( + store.clone(), + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Prepare, + mutation_id, + &intent.encode().expect("prepare intent should encode"), + ) + .await + .expect("prepare should persist the peer intent before commit contention"); + assert_eq!(prepared.state, TierMutationPeerState::Prepared); + assert!(prepared.applied); + let mut committed = intent.clone(); + committed + .advance(TierMutationIntentState::Committed, Some(config_etag.clone())) + .expect("matching intent should advance to committed"); + let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id) + .expect("intent record object should resolve"); + let barrier = crate::set_disk::PutObjectCommitBarrier::install( + RUSTFS_META_BUCKET, + &object, + crate::set_disk::PutObjectCommitPause::BeforeNamespace, + ); + let commit_store = store.clone(); + let commit_etag = config_etag.clone(); + let mut commit = tokio::spawn(async move { + handle_tier_mutation_peer_request( + commit_store, + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Commit, + mutation_id, + commit_etag.as_bytes(), + ) + .await + }); + barrier.wait_until_paused().await; + let namespace = store.pools[0].disk_set[0] + .new_ns_lock(RUSTFS_META_BUCKET, &object) + .await + .expect("intent record namespace should resolve"); + let owner = namespace + .get_write_lock(crate::set_disk::get_lock_acquire_timeout()) + .await + .expect("competing publisher should hold the intent namespace"); + // The competing publisher already owns the namespace. Bypass only its + // recursive acquisition and drain the durable tail before releasing it. + com::save_config_with_opts( + store.clone(), + &object, + committed.encode().expect("winning committed intent should encode"), + &ObjectOptions { + no_lock: true, + max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("competing publisher should persist the matching committed intent under its lock"); + let manager = ctx + .lock_manager() + .as_fast_lock_manager() + .expect("fixture should use real locks"); + let key = rustfs_lock::fast_lock::types::ObjectKey { + bucket: Arc::from(RUSTFS_META_BUCKET), + object: Arc::from(object.as_str()), + version: None, + }; + let timeouts = &manager.get_shard(&key).metrics().timeouts; + let before = timeouts.load(Ordering::Relaxed); + barrier.release_and_wait_until_namespace_pending().await; + drop(barrier); + tokio::select! { + result = &mut commit => panic!( + "Commit must retry the contended save and load before completing; observed {} lock timeouts: {result:?}", + timeouts.load(Ordering::Relaxed) - before, + ), + result = tokio::time::timeout(Duration::from_secs(30), async { + while timeouts.load(Ordering::Relaxed) - before < 2 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) => result.expect("the save and subsequent load should each exhaust their lock wait"), + } + assert_eq!(timeouts.load(Ordering::Relaxed) - before, 2); + drop(owner); + let outcome = tokio::time::timeout(Duration::from_secs(30), commit) + .await + .expect("third commit round should complete after the competing publisher releases its lock") + .expect("commit task should join") + .expect("third commit round should reload the matching durable committed intent"); + assert_eq!(outcome.state, TierMutationPeerState::Committed); + assert!( + !outcome.applied, + "the retry must reuse the winning committed record rather than rewrite it" + ); + assert_eq!( + load_tier_mutation_intent_record(store.clone(), mutation_id) + .await + .expect("winning committed record should remain readable"), + committed, + "the retry must preserve the full intent identity, candidate digest and committed config ETag", + ); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("committed tier config metadata should remain readable"); + assert_eq!(config_info.etag.as_deref(), Some(config_etag.as_str())); + assert_eq!(timeouts.load(Ordering::Relaxed) - before, 2); + shutdown.cancel(); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_mutation_peer_commit_preserves_lock_timeout_after_three_rounds() { + use crate::storage_api_contracts::namespace::NamespaceLocking as _; + use std::sync::atomic::Ordering; + + let temp_dir = tempfile::tempdir().expect("create exhausted tier commit store dir"); + let (ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-commit-exhaustion", &[4])).await; + register_mock_tier(&store.tier_config_mgr(), "COLD-A").await; + let (candidate_digest, config_etag) = { + let tier_config_mgr = store.tier_config_mgr(); + let manager = tier_config_mgr.read().await; + let candidate_digest = tier_config_candidate_digest(&manager).expect("peer commit candidate digest should build"); + manager + .save_tiering_config(store.clone()) + .await + .expect("peer commit config fixture should persist"); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("peer commit config fixture should load"); + (candidate_digest, config_info.etag.expect("peer commit config should carry an ETag")) + }; + let mutation_id = uuid::Uuid::new_v4(); + let mut intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", candidate_digest); + intent.old_config_etag = Some(config_etag.clone()); + let prepared = handle_tier_mutation_peer_request( + store.clone(), + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Prepare, + mutation_id, + &intent.encode().expect("prepare intent should encode"), + ) + .await + .expect("prepare should persist the peer intent before commit contention"); + assert_eq!(prepared.state, TierMutationPeerState::Prepared); + assert!(prepared.applied); + let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id) + .expect("intent record object should resolve"); + let barrier = crate::set_disk::PutObjectCommitBarrier::install( + RUSTFS_META_BUCKET, + &object, + crate::set_disk::PutObjectCommitPause::BeforeNamespace, + ); + let commit_store = store.clone(); + let commit_etag = config_etag.clone(); + let commit = tokio::spawn(async move { + handle_tier_mutation_peer_request( + commit_store, + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Commit, + mutation_id, + commit_etag.as_bytes(), + ) + .await + }); + barrier.wait_until_paused().await; + let namespace = store.pools[0].disk_set[0] + .new_ns_lock(RUSTFS_META_BUCKET, &object) + .await + .expect("intent record namespace should resolve"); + let lock_timeout = crate::set_disk::get_lock_acquire_timeout(); + let owner = namespace + .get_write_lock(lock_timeout) + .await + .expect("competing publisher should hold the intent namespace"); + let manager = ctx + .lock_manager() + .as_fast_lock_manager() + .expect("fixture should use real locks"); + let key = rustfs_lock::fast_lock::types::ObjectKey { + bucket: Arc::from(RUSTFS_META_BUCKET), + object: Arc::from(object.as_str()), + version: None, + }; + let timeouts = &manager.get_shard(&key).metrics().timeouts; + let before = timeouts.load(Ordering::Relaxed); + barrier.release_and_wait_until_namespace_pending().await; + drop(barrier); + let error = tokio::time::timeout(Duration::from_secs(30), commit) + .await + .expect("three contended commit rounds should exhaust the bounded budget") + .expect("commit task should join") + .expect_err("Commit must fail while the competing publisher keeps the namespace"); + match error { + TierMutationPeerError::Store(Error::Lock(rustfs_lock::LockError::Timeout { resource, timeout })) => { + assert_eq!(resource, key.to_string()); + assert_eq!(timeout, lock_timeout); + } + other => panic!("Commit must preserve the final typed lock timeout: {other:?}"), + } + let observed_timeouts = timeouts.load(Ordering::Relaxed) - before; + drop(owner); + assert_eq!( + load_tier_mutation_intent_record(store.clone(), mutation_id) + .await + .expect("failed commit must leave the prepared intent readable"), + intent, + "exhausted contention must preserve the full prepared identity, candidate digest and config ETag", + ); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("tier config metadata should remain readable after exhausted contention"); + assert_eq!(config_info.etag.as_deref(), Some(config_etag.as_str())); + assert_eq!(observed_timeouts, 3, "save and load must share three rounds"); + assert_eq!(timeouts.load(Ordering::Relaxed) - before, 3); + shutdown.cancel(); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_mutation_peer_commit_rejects_replaced_identity_after_lock_timeout() { + use crate::storage_api_contracts::namespace::NamespaceLocking as _; + use std::sync::atomic::Ordering; + + let temp_dir = tempfile::tempdir().expect("create conflicting tier commit store dir"); + let (ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-commit-identity-conflict", &[4])).await; + register_mock_tier(&store.tier_config_mgr(), "COLD-A").await; + let (candidate_digest, config_etag) = { + let tier_config_mgr = store.tier_config_mgr(); + let manager = tier_config_mgr.read().await; + let candidate_digest = tier_config_candidate_digest(&manager).expect("peer commit candidate digest should build"); + manager + .save_tiering_config(store.clone()) + .await + .expect("peer commit config fixture should persist"); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("peer commit config fixture should load"); + (candidate_digest, config_info.etag.expect("peer commit config should carry an ETag")) + }; + let mutation_id = uuid::Uuid::new_v4(); + let mut intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", candidate_digest); + intent.old_config_etag = Some(config_etag.clone()); + let prepared = handle_tier_mutation_peer_request( + store.clone(), + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Prepare, + mutation_id, + &intent.encode().expect("prepare intent should encode"), + ) + .await + .expect("prepare should persist the peer intent before commit contention"); + assert_eq!(prepared.state, TierMutationPeerState::Prepared); + assert!(prepared.applied); + let mut replacement = intent.clone(); + replacement.expires_at_unix_nanos = replacement + .expires_at_unix_nanos + .checked_sub(1) + .expect("replacement expiry should fit i64"); + replacement + .advance(TierMutationIntentState::Committed, Some(config_etag.clone())) + .expect("conflicting identity should still form a valid committed record"); + assert!(!replacement.same_identity_as(&intent)); + let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id) + .expect("intent record object should resolve"); + let barrier = crate::set_disk::PutObjectCommitBarrier::install( + RUSTFS_META_BUCKET, + &object, + crate::set_disk::PutObjectCommitPause::BeforeNamespace, + ); + let commit_store = store.clone(); + let commit_etag = config_etag.clone(); + let commit = tokio::spawn(async move { + handle_tier_mutation_peer_request( + commit_store, + TIER_MUTATION_RPC_PROTOCOL_VERSION, + TierMutationRpcPhase::Commit, + mutation_id, + commit_etag.as_bytes(), + ) + .await + }); + barrier.wait_until_paused().await; + let namespace = store.pools[0].disk_set[0] + .new_ns_lock(RUSTFS_META_BUCKET, &object) + .await + .expect("intent record namespace should resolve"); + let owner = namespace + .get_write_lock(crate::set_disk::get_lock_acquire_timeout()) + .await + .expect("competing publisher should hold the intent namespace"); + // The publisher owns the namespace, so only its recursive acquisition + // is bypassed while the conflicting record reaches durable completion. + com::save_config_with_opts( + store.clone(), + &object, + replacement.encode().expect("conflicting committed intent should encode"), + &ObjectOptions { + no_lock: true, + max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("competing publisher should persist the conflicting intent under its lock"); + let manager = ctx + .lock_manager() + .as_fast_lock_manager() + .expect("fixture should use real locks"); + let key = rustfs_lock::fast_lock::types::ObjectKey { + bucket: Arc::from(RUSTFS_META_BUCKET), + object: Arc::from(object.as_str()), + version: None, + }; + let timeouts = &manager.get_shard(&key).metrics().timeouts; + let before = timeouts.load(Ordering::Relaxed); + barrier.release_and_wait_until_namespace_pending().await; + drop(barrier); + tokio::time::timeout(Duration::from_secs(30), async { + while timeouts.load(Ordering::Relaxed) == before { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the conditional save should exhaust its lock wait before ownership is released"); + assert_eq!(timeouts.load(Ordering::Relaxed) - before, 1); + drop(owner); + let error = tokio::time::timeout(Duration::from_secs(30), commit) + .await + .expect("commit should finish after the conflicting publisher releases its lock") + .expect("commit task should join") + .expect_err("commit must reject the identity replaced during its lock timeout"); + assert_eq!( + load_tier_mutation_intent_record(store.clone(), mutation_id) + .await + .expect("conflicting committed record should remain readable"), + replacement, + "rejected commit must not rewrite the competing identity", + ); + let config_info = store + .get_object_info( + RUSTFS_META_BUCKET, + &format!("{}/{}", com::CONFIG_PREFIX, TIER_CONFIG_FILE), + &ObjectOptions::default(), + ) + .await + .expect("tier config metadata should remain readable after rejecting the conflict"); + assert_eq!(config_info.etag.as_deref(), Some(config_etag.as_str())); + assert_eq!(timeouts.load(Ordering::Relaxed) - before, 1); + assert!( + matches!(error, TierMutationPeerError::Store(Error::PreconditionFailed)), + "post-timeout reload must reject the replaced identity with PreconditionFailed: {error:?}" + ); + shutdown.cancel(); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)]