From d386f7741dc34017c0e40f95e079f92361c25eda Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 8 Oct 2026 21:24:34 +0800 Subject: [PATCH] fix(lifecycle): coordinate tier free-version recovery scans (#8387) * fix(lifecycle): coordinate tier free-version recovery scans * fix(lifecycle): remove obsolete recovery enqueue entry point * test(upgrade): wait for writers after each rolling replacement --- .../e2e_test/src/distributed/upgrade_test.rs | 20 +- .../src/upgrade_compatibility_test.rs | 42 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 409 ++++++++++++++++-- .../lifecycle/tier_free_version_recovery.rs | 8 +- docs/operations/tier-ilm-debugging.md | 8 +- 5 files changed, 437 insertions(+), 50 deletions(-) diff --git a/crates/e2e_test/src/distributed/upgrade_test.rs b/crates/e2e_test/src/distributed/upgrade_test.rs index d29c6f61f..4d9af12f0 100644 --- a/crates/e2e_test/src/distributed/upgrade_test.rs +++ b/crates/e2e_test/src/distributed/upgrade_test.rs @@ -31,6 +31,7 @@ use super::harness::{ use crate::common::{ AdminTransport, admin_add_canned_policy_via, admin_attach_user_policy_via, admin_create_user_via, init_logging, }; +use crate::upgrade_compatibility_test::wait_for_upgrade_write_readiness; use aws_sdk_s3::Client; use aws_sdk_s3::error::ProvideErrorMetadata; use std::path::{Path, PathBuf}; @@ -400,6 +401,19 @@ async fn four_node_direct_upgrade_preserves_history_and_iam_credentials() -> Tes Ok(()) } +async fn replace_node_and_wait_for_write_readiness(dist: &mut DistCluster, seed: &UpgradeSeed, node_idx: usize) -> TestResult { + dist.replace_node_with_current_binary(node_idx).await?; + // Node readiness proves reads can succeed; a rolling writer must also + // recover before replacing the next peer or checking upgrade contracts. + wait_for_upgrade_write_readiness( + &dist.clients()?, + &seed.history_bucket, + &format!("replaced-node-{node_idx}"), + CREDENTIAL_TIMEOUT, + ) + .await +} + #[tokio::test] async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> TestResult { init_logging(); @@ -410,15 +424,15 @@ async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> Te let seed = seed_history_and_iam(&dist).await?; - dist.replace_node_with_current_binary(0).await?; + replace_node_and_wait_for_write_readiness(&mut dist, &seed, 0).await?; assert_history_and_iam(&dist, &seed, "one-current-node").await?; for node_idx in [1, 2] { - dist.replace_node_with_current_binary(node_idx).await?; + replace_node_and_wait_for_write_readiness(&mut dist, &seed, node_idx).await?; } assert_history_and_iam(&dist, &seed, "one-previous-node").await?; - dist.replace_node_with_current_binary(3).await?; + replace_node_and_wait_for_write_readiness(&mut dist, &seed, 3).await?; assert_history_and_iam(&dist, &seed, "homogeneous-current").await?; Ok(()) } diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index d9e825766..51ff93d6f 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -578,7 +578,7 @@ async fn exercise_mixed_cluster( previous_node: usize, ) -> TestResult { let clients = cluster.create_all_clients()?; - wait_for_upgrade_write_readiness(&clients, phase, LISTING_CONVERGENCE_TIMEOUT).await?; + wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, phase, LISTING_CONVERGENCE_TIMEOUT).await?; let current_client = &clients[current_node]; let previous_client = &clients[previous_node]; @@ -646,7 +646,12 @@ fn upgrade_probe_client(client: &Client) -> Client { ) } -async fn wait_for_upgrade_write_readiness(clients: &[Client], phase: &str, budget: Duration) -> TestResult { +pub(crate) async fn wait_for_upgrade_write_readiness( + clients: &[Client], + bucket: &str, + phase: &str, + budget: Duration, +) -> TestResult { // ListBuckets can succeed before peers recover a restarted disk. Cluster // health also accepts Returning disks whose write health is still FAULTY. // Probe every writer outside the asserted phase prefix; compatibility @@ -666,7 +671,7 @@ async fn wait_for_upgrade_write_readiness(clients: &[Client], phase: &str, budge deadline, client .put_object() - .bucket(MIXED_BUCKET) + .bucket(bucket) .key(&key) .body(ByteStream::from_static(UPGRADE_READINESS_BODY)) .send(), @@ -701,19 +706,20 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro // waits for pool metadata (#7473). First prove the elected writer can // persist data, then restart each old process once with three peers still // readable. This preparation ends before any current binary is started. - wait_for_upgrade_write_readiness(&clients[..1], "previous-seed", LISTING_CONVERGENCE_TIMEOUT).await?; + wait_for_upgrade_write_readiness(&clients[..1], MIXED_BUCKET, "previous-seed", LISTING_CONVERGENCE_TIMEOUT).await?; for node in [1, 2, 3, 0] { cluster.stop_node(node)?; cluster.start_node_from_binary(node, previous_binary).await?; wait_for_upgrade_write_readiness( std::slice::from_ref(&clients[node]), + MIXED_BUCKET, &format!("previous-restart-{node}"), LISTING_CONVERGENCE_TIMEOUT, ) .await?; } - wait_for_upgrade_write_readiness(&clients, "previous-baseline", LISTING_CONVERGENCE_TIMEOUT).await?; + wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, "previous-baseline", LISTING_CONVERGENCE_TIMEOUT).await?; for (reader, client) in clients.iter().enumerate() { assert_eq!( read_object(client, MIXED_BUCKET, PREVIOUS_RELEASE_SEED_KEY, None).await?.1, @@ -740,9 +746,10 @@ mod upgrade_write_readiness_tests { #[tokio::test] async fn waits_for_each_writer_after_metadata_is_ready() -> TestResult { let target = FakeS3Target::start().await?; - target.create_bucket(MIXED_BUCKET); + let bucket = "distributed-upgrade-history"; + target.create_bucket(bucket); let client = fake_source_client(&target); - client.head_bucket().bucket(MIXED_BUCKET).send().await?; + client.head_bucket().bucket(bucket).send().await?; let phase = "one-previous-node"; let first_key = format!(".upgrade-readiness/{phase}/node-0"); @@ -762,7 +769,7 @@ mod upgrade_write_readiness_tests { // Metadata readiness does not prove that a data write can succeed. let premature = client .put_object() - .bucket(MIXED_BUCKET) + .bucket(bucket) .key(&first_key) .body(ByteStream::from_static(b"upgrade write readiness")) .send() @@ -770,15 +777,15 @@ mod upgrade_write_readiness_tests { .expect_err("metadata readiness does not prove write readiness"); assert_eq!(premature.raw_response().map(|response| response.status().as_u16()), Some(503)); - wait_for_upgrade_write_readiness(&[client.clone(), client.clone()], phase, Duration::from_secs(5)).await?; + wait_for_upgrade_write_readiness(&[client.clone(), client.clone()], bucket, phase, Duration::from_secs(5)).await?; assert_eq!(target.count_requests(FakeTargetOperation::PutObject, &first_key), 3); assert_eq!(target.count_requests(FakeTargetOperation::PutObject, &second_key), 2); - assert!(target.has_object(MIXED_BUCKET, &first_key)); - assert!(target.has_object(MIXED_BUCKET, &second_key)); + assert!(target.has_object(bucket, &first_key)); + assert!(target.has_object(bucket, &second_key)); assert!( client .list_objects_v2() - .bucket(MIXED_BUCKET) + .bucket(bucket) .prefix(format!("{phase}/")) .send() .await? @@ -808,9 +815,10 @@ mod upgrade_write_readiness_tests { let phase = format!("permanent-{}", status.as_u16()); let key = format!(".upgrade-readiness/{phase}/node-0"); target.inject_for_key(FakeTargetOperation::PutObject, &key, FaultAction::Status(status), 1); - let error = wait_for_upgrade_write_readiness(std::slice::from_ref(&client), &phase, Duration::from_secs(5)) - .await - .expect_err("a permanent error must not be retried into success"); + let error = + wait_for_upgrade_write_readiness(std::slice::from_ref(&client), MIXED_BUCKET, &phase, Duration::from_secs(5)) + .await + .expect_err("a permanent error must not be retried into success"); assert!(error.to_string().contains("node 0 write readiness failed"), "{error}"); assert_eq!(target.count_requests(FakeTargetOperation::PutObject, &key), 1); assert!(!target.has_object(MIXED_BUCKET, &key)); @@ -833,7 +841,7 @@ mod upgrade_write_readiness_tests { FaultAction::Status(StatusCode::SERVICE_UNAVAILABLE), 10, ); - let error = wait_for_upgrade_write_readiness(&[client], phase, Duration::from_secs(1)) + let error = wait_for_upgrade_write_readiness(&[client], MIXED_BUCKET, phase, Duration::from_secs(1)) .await .expect_err("persistent unavailability must exhaust the shared deadline"); // Transport scheduling consumes the same budget; do not require a @@ -852,7 +860,7 @@ mod upgrade_write_readiness_tests { let phase = "stalled"; let key = format!(".upgrade-readiness/{phase}/node-0"); target.inject_for_key(FakeTargetOperation::PutObject, &key, FaultAction::Stall(Duration::from_secs(30)), 1); - let error = wait_for_upgrade_write_readiness(std::slice::from_ref(&client), phase, Duration::from_secs(1)) + let error = wait_for_upgrade_write_readiness(std::slice::from_ref(&client), MIXED_BUCKET, phase, Duration::from_secs(1)) .await .expect_err("a stalled request must not outlive the readiness deadline"); assert!(error.to_string().contains("deadline exceeded during PutObject"), "{error}"); diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index f1aa911bc..1d0d9ae48 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -87,6 +87,7 @@ use rustfs_filemeta::{ FileInfo, FileInfoOpts, NULL_VERSION_ID, RestoreStatus, RestoreStatusOps, TRANSITION_COMPLETE, get_file_info, is_restored_object_on_disk, }; +use rustfs_lock::{NamespaceLockGuard, error::LockError}; use rustfs_scanner_metrics::metrics::{ IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics, }; @@ -190,6 +191,8 @@ const TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL: StdDuration = StdDuration::from_ // before another full sweep can discover work persisted by another node. const TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL: StdDuration = StdDuration::from_secs(10 * 60); const TIER_FREE_VERSION_RECOVERY_JITTER_PERCENT: u64 = 10; +const TIER_FREE_VERSION_RECOVERY_LEADER_LOCK: &str = "tier-free-version-recovery/leader.lock"; +const TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT: StdDuration = StdDuration::from_secs(1); const DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS: i64 = 5; const EXPIRY_WORKER_QUEUE_CAPACITY: usize = 1000; const DEFAULT_MANUAL_TRANSITION_JOB_RECOVERY_LIMIT: usize = 100; @@ -1441,13 +1444,24 @@ pub(crate) async fn enqueue_committed_free_versions(api: &ECStore, free_versions queued } -async fn enqueue_recovered_free_version_with_state(state: &Arc>, oi: ObjectInfo) -> bool { +async fn enqueue_recovered_free_version_with_state( + state: &Arc>, + oi: ObjectInfo, + cancel: &CancellationToken, +) -> bool { let task = FreeVersionTask(oi); let hash = task.op_hash(); let (wrkr, stats) = { - let state = state.read().await; + let state = select! { + biased; + _ = cancel.cancelled() => return false, + state = state.read() => state, + }; (state.get_worker_ch(hash), Arc::clone(&state.stats)) }; + if cancel.is_cancelled() { + return false; + } let Some(wrkr) = wrkr else { stats.increment_missed_freevers_tasks(); stats.record_scanner_expiry_state(); @@ -1497,9 +1511,13 @@ fn set_recovered_free_version_enqueue_observer( RecoveredFreeVersionEnqueueObserverGuard } -pub async fn enqueue_recovered_free_version(api: &ECStore, oi: ObjectInfo) -> bool { +pub(super) async fn enqueue_recovered_free_version_with_cancel( + api: &ECStore, + oi: ObjectInfo, + cancel: &CancellationToken, +) -> bool { let expiry_state = api.ctx.expiry_state(); - let queued = enqueue_recovered_free_version_with_state(&expiry_state, oi).await; + let queued = enqueue_recovered_free_version_with_state(&expiry_state, oi, cancel).await; #[cfg(test)] if let Some(observer) = RECOVERED_FREE_VERSION_ENQUEUE_OBSERVER @@ -2937,20 +2955,43 @@ fn spawn_tier_free_version_recovery_once(api: Arc, started: &OnceLock<( Some(tokio::spawn(async move { let cancel_token = api.ctx.background_cancel_token().unwrap_or_default(); - let expiry_state = api.ctx.expiry_state(); - run_tier_free_version_recovery_loop( + let acquire_api = Arc::clone(&api); + run_tier_free_version_recovery_coordinator( cancel_token, - expiry_state, - jitter_tier_free_version_recovery_delay, - move |bucket_marker, object_marker, recovery_cancel| { + tier_free_version_recovery_leader_retry_delay, + move || { + let api = Arc::clone(&acquire_api); + async move { + let lock = api + .new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK) + .await?; + match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await { + Ok(guard) => Ok(Some(guard)), + Err(LockError::Timeout { .. }) => Ok(None), + Err(err) => Err(err.into()), + } + } + }, + move |leader_cancel| { let api = Arc::clone(&api); async move { - recover_tier_free_versions_with_cancel( - api, - DEFAULT_FREE_VERSION_RECOVERY_LIMIT, - bucket_marker, - object_marker, - recovery_cancel, + run_tier_free_version_recovery_loop( + leader_cancel, + api.ctx.expiry_state(), + jitter_tier_free_version_recovery_delay, + move |bucket_marker, object_marker, recovery_cancel| { + let api = Arc::clone(&api); + async move { + recover_tier_free_versions_with_cancel( + api, + DEFAULT_FREE_VERSION_RECOVERY_LIMIT, + bucket_marker, + object_marker, + recovery_cancel, + ) + .await + } + }, ) .await } @@ -2960,6 +3001,86 @@ fn spawn_tier_free_version_recovery_once(api: Arc, started: &OnceLock<( })) } +fn tier_free_version_recovery_leader_retry_delay() -> StdDuration { + StdDuration::from_secs(rand::rng().random_range(15..=30)) +} + +async fn run_tier_free_version_recovery_coordinator( + cancel: CancellationToken, + retry_delay: fn() -> StdDuration, + mut acquire: A, + mut recover: R, +) where + A: FnMut() -> AF, + AF: Future>>, + R: FnMut(CancellationToken) -> RF, + RF: Future, +{ + loop { + let acquired = select! { + biased; + _ = cancel.cancelled() => return, + result = acquire() => result, + }; + match acquired { + Ok(Some(mut guard)) => { + let leader_cancel = cancel.child_token(); + // Hold ownership across every page, cooldown and idle wait. A + // follower taking over starts a fresh sweep of persisted markers. + let recovery = recover(leader_cancel.clone()); + tokio::pin!(recovery); + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "free_version_recovery_leader_acquired", + lock_name = TIER_FREE_VERSION_RECOVERY_LEADER_LOCK, + "Tier free-version recovery leader acquired" + ); + select! { + biased; + _ = cancel.cancelled() => { + leader_cancel.cancel(); + recovery.await; + } + _ = guard.lock_lost_notified() => { + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "free_version_recovery_leader_lost", + lock_name = TIER_FREE_VERSION_RECOVERY_LEADER_LOCK, + "Tier free-version recovery leader lost" + ); + // Dropping the recovery future would detach its active + // walk. Cancel it and poll the same future through cleanup. + leader_cancel.cancel(); + recovery.await; + } + _ = &mut recovery => {} + } + leader_cancel.cancel(); + guard.release(); + } + Ok(None) => {} + Err(err) => warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "free_version_recovery_leader_acquire_failed", + lock_name = TIER_FREE_VERSION_RECOVERY_LEADER_LOCK, + error = %err, + "Tier free-version recovery leader acquisition failed" + ), + } + select! { + biased; + _ = cancel.cancelled() => return, + _ = tokio::time::sleep(retry_delay()) => {} + } + } +} + async fn run_tier_free_version_recovery_loop( cancel_token: CancellationToken, expiry_state: Arc>, @@ -2969,7 +3090,11 @@ async fn run_tier_free_version_recovery_loop( F: FnMut(Option, Option, CancellationToken) -> Fut, Fut: Future>, { - let recovery_notify = Arc::clone(&expiry_state.read().await.recovery_notify); + let recovery_notify = select! { + biased; + _ = cancel_token.cancelled() => return, + state = expiry_state.read() => Arc::clone(&state.recovery_notify), + }; let mut schedule = TierFreeVersionRecoverySchedule::default(); loop { @@ -2992,7 +3117,11 @@ async fn run_tier_free_version_recovery_loop( schedule.record_success(&stats); rustfs_io_metrics::record_stage_duration("lifecycle_free_version_recovery", elapsed.as_secs_f64() * 1000.0); let (pending_tasks, active_tasks) = { - let state = expiry_state.read().await; + let state = select! { + biased; + _ = cancel_token.cancelled() => return, + state = expiry_state.read() => state, + }; (state.pending_tasks(), state.stats.active_tasks()) }; debug!( @@ -5906,10 +6035,10 @@ mod tests { merge_stale_multipart_candidate, persist_manual_transition_job_progress_if_owned, persist_manual_transition_page_checkpoint, recover_manual_transition_job, recover_manual_transition_jobs, resolve_tier_free_version_recovery_enabled, resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, - resolve_transition_worker_count, resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, - select_restore_s3_location, set_lifecycle_observability_observer, set_recovered_free_version_enqueue_observer, - should_defer_date_expiry_for_recent_config_update, transitioned_cleanup_tuple, transitioned_object_delete_opts, - wait_for_tier_free_version_recovery, + resolve_transition_worker_count, resolve_transition_workers_absolute_max, run_tier_free_version_recovery_coordinator, + run_tier_free_version_recovery_loop, select_restore_s3_location, set_lifecycle_observability_observer, + set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update, + transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery, }; #[cfg(feature = "test-util")] use super::{delete_free_version_remote_object_then, encode_dir_object, get_transitioned_object_reader_with_tier_manager}; @@ -6003,6 +6132,231 @@ mod tests { use tokio_util::sync::CancellationToken; use uuid::Uuid; + fn recovery_test_lock() -> (Arc, Vec>) { + let clients: Vec> = (0..3) + .map(|_| { + Arc::new(rustfs_lock::LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))) + as Arc + }) + .collect(); + let lock = + rustfs_lock::NamespaceLock::with_clients_and_quorum(format!("recovery-test-{}", Uuid::new_v4()), clients.clone(), 2); + (Arc::new(lock), clients) + } + + async fn acquire_recovery_test_lock( + lock: Arc, + owner: String, + ) -> crate::error::Result> { + // Keep this lease longer than the virtual-time cooldown/idle advances; + // production uses the namespace wrapper's renewable 30-second lease. + lock.lock_guard( + rustfs_lock::ObjectKey::new(crate::disk::RUSTFS_META_BUCKET, super::TIER_FREE_VERSION_RECOVERY_LEADER_LOCK), + &owner, + StdDuration::from_millis(5), + StdDuration::from_secs(3600), + ) + .await + .map_err(Into::into) + } + + fn recovery_test_retry_delay() -> StdDuration { + StdDuration::from_secs(15) + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn tier_free_version_recovery_eight_coordinators_hold_one_leader_and_take_over() { + let (lock, _clients) = recovery_test_lock(); + let (pages_tx, mut pages_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut tasks = Vec::new(); + let mut cancels = Vec::new(); + for node in 0..8 { + let cancel = CancellationToken::new(); + cancels.push(cancel.clone()); + let lock = Arc::clone(&lock); + let pages_tx = pages_tx.clone(); + tasks.push(tokio::spawn(async move { + run_tier_free_version_recovery_coordinator( + cancel, + recovery_test_retry_delay, + move || acquire_recovery_test_lock(Arc::clone(&lock), format!("node-{node}")), + move |leader_cancel| { + let pages_tx = pages_tx.clone(); + async move { + let mut pages = 0; + run_tier_free_version_recovery_loop( + leader_cancel, + ExpiryState::new(), + std::convert::identity, + move |bucket, object, _| { + pages += 1; + pages_tx.send((node, bucket, object)).expect("page observer should be open"); + std::future::ready(Ok(free_version_recovery_stats(0, 0, pages == 1))) + }, + ) + .await; + } + }, + ) + .await; + })); + } + let (leader, bucket, object) = pages_rx.recv().await.expect("one coordinator should start recovery"); + assert!(bucket.is_none() && object.is_none()); + // The same leader must survive both a truncated page's cooldown and + // complete idle sweeps. All other coordinators keep probing the lock. + let mut observed_pages = 1; + for _ in 0..27 { + tokio::time::advance(StdDuration::from_secs(60)).await; + for _ in 0..100 { + tokio::task::yield_now().await; + } + while let Ok((node, bucket, object)) = pages_rx.try_recv() { + assert_eq!(node, leader, "followers must not repeat a namespace sweep"); + if observed_pages == 1 { + assert_eq!(bucket.as_deref(), Some("bucket")); + assert_eq!(object.as_deref(), Some("object")); + } else { + assert!(bucket.is_none() && object.is_none(), "complete sweeps should reset their cursors"); + } + observed_pages += 1; + } + } + assert!(observed_pages >= 7, "the leader should finish a full ten-minute idle wait"); + cancels[leader].cancel(); + let (successor, bucket, object) = tokio::time::timeout(StdDuration::from_secs(60), pages_rx.recv()) + .await + .expect("a follower should take over after leader shutdown") + .expect("page observer should remain open"); + assert_ne!(successor, leader); + assert!(bucket.is_none() && object.is_none(), "a new leader must restart discovery safely"); + for cancel in cancels { + cancel.cancel(); + } + for task in tasks { + task.await.expect("all coordinators should shut down cleanly"); + } + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn tier_free_version_recovery_lock_loss_waits_for_active_page_cleanup() { + let (lock, _clients) = recovery_test_lock(); + let cancel = CancellationToken::new(); + let loop_cancel = cancel.clone(); + let (page_tx, mut page_rx) = tokio::sync::mpsc::unbounded_channel(); + let cleanup_started = Arc::new(tokio::sync::Notify::new()); + let finish_cleanup = Arc::new(tokio::sync::Notify::new()); + let observed_cleanup = Arc::clone(&cleanup_started); + let release_cleanup = Arc::clone(&finish_cleanup); + let task = tokio::spawn(async move { + run_tier_free_version_recovery_coordinator( + loop_cancel, + recovery_test_retry_delay, + move || acquire_recovery_test_lock(Arc::clone(&lock), "old-leader".to_string()), + move |leader_cancel| { + let page_tx = page_tx.clone(); + let cleanup_started = Arc::clone(&cleanup_started); + let finish_cleanup = Arc::clone(&finish_cleanup); + async move { + run_tier_free_version_recovery_loop( + leader_cancel, + ExpiryState::new(), + std::convert::identity, + move |_, _, page_cancel| { + let cleanup_started = Arc::clone(&cleanup_started); + let finish_cleanup = Arc::clone(&finish_cleanup); + page_tx.send(()).expect("page observer should remain open"); + async move { + page_cancel.cancelled().await; + cleanup_started.notify_one(); + finish_cleanup.notified().await; + Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "cancelled page").into()) + } + }, + ) + .await; + } + }, + ) + .await; + }); + page_rx.recv().await.expect("leader should start an active page"); + // Let authoritative backend leases expire before the next heartbeat + // can renew them. A failed refresh must cancel the active page. + tokio::time::advance(StdDuration::from_secs(3600)).await; + tokio::time::timeout(StdDuration::from_secs(1), observed_cleanup.notified()) + .await + .expect("expired backend leases must trigger recovery cancellation"); + assert!(!task.is_finished(), "lock loss must wait for the active page's cleanup"); + assert!(page_rx.try_recv().is_err(), "the old leader must not start another page"); + cancel.cancel(); + release_cleanup.notify_one(); + task.await.expect("coordinator should finish cleanup before exiting"); + } + + #[tokio::test(start_paused = true)] + async fn tier_free_version_recovery_coordinator_retries_without_scanning_after_lock_error() { + let cancel = CancellationToken::new(); + let loop_cancel = cancel.clone(); + let attempts = Arc::new(AtomicUsize::new(0)); + let observed_attempts = Arc::clone(&attempts); + let task = tokio::spawn(async move { + run_tier_free_version_recovery_coordinator( + loop_cancel, + recovery_test_retry_delay, + move || { + observed_attempts.fetch_add(1, Ordering::SeqCst); + std::future::ready(Err(rustfs_lock::LockError::InsufficientNodes { + required: 2, + available: 1, + } + .into())) + }, + |_| async { panic!("a node without ownership must never start discovery") }, + ) + .await; + }); + tokio::task::yield_now().await; + assert_eq!(attempts.load(Ordering::SeqCst), 1); + tokio::time::advance(StdDuration::from_secs(14)).await; + assert_eq!(attempts.load(Ordering::SeqCst), 1, "lock errors must not cause a tight retry loop"); + tokio::time::advance(StdDuration::from_secs(1)).await; + tokio::task::yield_now().await; + assert_eq!(attempts.load(Ordering::SeqCst), 2); + cancel.cancel(); + task.await.expect("a follower should respond to shutdown during retry wait"); + } + + #[tokio::test] + async fn tier_free_version_recovery_enqueue_cancels_while_waiting_for_state_lock() { + let state = ExpiryState::new(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1); + let mut held = state.write().await; + held.tasks_tx.push(tx); + let cancel = CancellationToken::new(); + let enqueue_cancel = cancel.clone(); + let enqueue_state = Arc::clone(&state); + let enqueue = tokio::spawn(async move { + enqueue_recovered_free_version_with_state(&enqueue_state, ObjectInfo::default(), &enqueue_cancel).await + }); + tokio::task::yield_now().await; + assert!(!enqueue.is_finished(), "enqueue should be blocked on the state lock"); + cancel.cancel(); + let queued = tokio::time::timeout(StdDuration::from_secs(1), enqueue) + .await + .expect("cancellation must not wait for the state write guard") + .expect("enqueue should return cleanly"); + assert!(!queued); + assert_eq!(held.pending_tasks(), 0); + assert_eq!(held.stats.missed_free_vers_tasks(), 0); + assert!(rx.try_recv().is_err(), "a cancelled recovery must not enqueue a task"); + drop(held); + assert!(!enqueue_recovered_free_version_with_state(&state, ObjectInfo::default(), &cancel).await); + assert!(rx.try_recv().is_err(), "cancellation must win even when the lock is available"); + } + fn free_version_recovery_stats(enqueued: usize, failed: usize, truncated: bool) -> FreeVersionRecoveryStats { FreeVersionRecoveryStats { scanned: enqueued.saturating_add(failed), @@ -7682,7 +8036,7 @@ mod tests { assert_eq!(state.read().await.stats.pending_tasks(), 1); // The single-slot queue is full for both enqueue paths. - assert!(!enqueue_recovered_free_version_with_state(&state, oi.clone()).await); + assert!(!enqueue_recovered_free_version_with_state(&state, oi.clone(), &CancellationToken::new()).await); assert_eq!(state.read().await.stats.pending_tasks(), 1); assert_eq!(state.read().await.stats.missed_free_vers_tasks(), 1); @@ -7714,7 +8068,7 @@ mod tests { ..Default::default() }; - let queued = enqueue_recovered_free_version_with_state(&state, oi).await; + let queued = enqueue_recovered_free_version_with_state(&state, oi, &CancellationToken::new()).await; let state = state.read().await; assert!(!queued); @@ -7843,8 +8197,8 @@ mod tests { ..Default::default() }; - let first = enqueue_recovered_free_version_with_state(&state, oi.clone()).await; - let second = enqueue_recovered_free_version_with_state(&state, oi).await; + let first = enqueue_recovered_free_version_with_state(&state, oi.clone(), &CancellationToken::new()).await; + let second = enqueue_recovered_free_version_with_state(&state, oi, &CancellationToken::new()).await; let state = state.read().await; assert!(first); @@ -7945,7 +8299,7 @@ mod tests { }; assert!( - super::enqueue_recovered_free_version(&ecstore, oi).await, + super::enqueue_recovered_free_version_with_cancel(&ecstore, oi, &CancellationToken::new()).await, "the resized production worker queue should accept the task" ); stop_tx.send(None).await.expect("worker stop signal should be delivered"); @@ -13898,13 +14252,14 @@ mod tests { }; let mut recovery_rx = recovery_rx.lock().await; assert!( - super::enqueue_recovered_free_version( + super::enqueue_recovered_free_version_with_cancel( &ecstore, ObjectInfo { bucket: "prefill".to_string(), name: "prefill".to_string(), ..Default::default() }, + &CancellationToken::new(), ) .await, "the production recovery queue should accept its first task" diff --git a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs index 4ccec6f23..27f767846 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs @@ -20,7 +20,7 @@ use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -use crate::bucket::lifecycle::bucket_lifecycle_ops::enqueue_recovered_free_version; +use crate::bucket::lifecycle::bucket_lifecycle_ops::enqueue_recovered_free_version_with_cancel; use crate::disk::RUSTFS_META_BUCKET; use crate::error::Result; use crate::object_api::ObjectInfo; @@ -201,7 +201,11 @@ pub(super) async fn recover_tier_free_versions_with_cancel( return Err(tier_free_version_recovery_cancelled()); } retry_cursor.visit(&oi); - if !record_recovered_free_version_enqueue(&mut stats, enqueue_recovered_free_version(&api, oi).await) { + let queued = enqueue_recovered_free_version_with_cancel(&api, oi, &cancel_token).await; + if cancel_token.is_cancelled() { + return Err(tier_free_version_recovery_cancelled()); + } + if !record_recovered_free_version_enqueue(&mut stats, queued) { let (bucket_marker, object_marker) = retry_cursor.retry_markers(); stats.truncated = true; stats.next_bucket_marker = bucket_marker; diff --git a/docs/operations/tier-ilm-debugging.md b/docs/operations/tier-ilm-debugging.md index 6c91e391b..a9a1e05a1 100644 --- a/docs/operations/tier-ilm-debugging.md +++ b/docs/operations/tier-ilm-debugging.md @@ -36,11 +36,17 @@ An expired restored local copy can be cleaned up under Object Lock because the r The dedicated free-version recovery loop is enabled by default and is independent of the data scanner and heal switches. Setting `RUSTFS_SCANNER_ENABLED=false` does not stop this repair loop. Set `RUSTFS_TIER_FREE_VERSION_RECOVERY_ENABLED=false` before process startup to disable only the dedicated persisted-marker walk. That setting does not disable lifecycle workers or prevent another scanner path from discovering a free version, and it can leave remote cleanup markers pending for longer, so use it as a break-glass pressure control rather than a cleanup mechanism. +Enabled nodes compete for the independent `tier-free-version-recovery/leader.lock` namespace lock in the internal metadata bucket. The leader holds the renewable lock across pages, cooldowns, retries, and idle waits; followers only attempt ownership with a one-second acquisition budget and a random 15–30 second delay between attempts. This uses the existing pool 0 locker majority, so recovery pauses when that lock domain cannot form quorum even if another pool remains available. Recovery resumes when ownership can be acquired again. It does not depend on the main scanner's leader lock or enable switch. + +Loss of the leader lease or process shutdown cancels discovery and waits for active walks to stop before the coordinator exits that leadership term. A successor starts a new sweep from the beginning because continuation markers remain process-local; frequent leadership changes can therefore delay discovery near the end of a large namespace. Already queued cleanup tasks retain their existing exact-version and object-lock checks and may finish after leadership changes. During a rolling upgrade, older nodes still perform independent walks until they are upgraded or recovery is disabled on them. + +Recovery discoveries enter the leader's local lifecycle worker queues. Queue pressure preserves the failed object's retry position; spare workers on other nodes do not automatically take those tasks. Recovery notifications are also process-local: a failed enqueue or cleanup on a follower is rediscovered through the leader's periodic sweeps rather than immediately waking it. The ten-minute idle cap bounds the idle gap, not the total time to discover a marker, which also includes page cooldowns and traversal time. Persisted free-version markers remain the retry source of truth. + Normal transitioned deletes pass a post-commit receipt directly to the lifecycle queue. The namespace walk is the crash, queue-pressure, mixed-version, and historical-record fallback. It cannot safely skip a bucket merely because that bucket has no current lifecycle rule or the referenced tier was removed: an older `xl.meta` free-version can still be the only owner of a required remote DELETE. One background recovery page completes at most 10,000 logical objects across at most 100 buckets and enqueues at most 1,000 recoverable free versions. The scanner can decode one additional object to detect truncation; a continuation marker preserves the first unscanned bucket or the last completely scanned object. Every truncated page and follow-up sweep waits at least 60 seconds after the previous page completes; failed pages back off from 60 seconds to 10 minutes, and complete idle sweeps exponentially back off to 10 minutes with jitter. Individual walks have no fixed total timeout, but inherit the drive walk stall timeout so a large healthy bucket can make progress without recreating the old timeout/restart loop. -The structured `lifecycle_worker_state` recovery event reports `duration_ms`, `scanned_entries`, `buckets_scanned`, queue counts, truncation, and continuation markers. `rustfs_internal_stage_duration_ms{stage="lifecycle_free_version_recovery"}` records successful page duration; `stage="lifecycle_free_version_recovery_failed"` records failures. +The structured `lifecycle_worker_state` recovery event reports `duration_ms`, `scanned_entries`, `buckets_scanned`, queue counts, truncation, and continuation markers. Coordinator states include `free_version_recovery_leader_acquired`, `free_version_recovery_leader_lost`, and `free_version_recovery_leader_acquire_failed`; normal follower contention is quiet. `rustfs_internal_stage_duration_ms{stage="lifecycle_free_version_recovery"}` records successful page duration; `stage="lifecycle_free_version_recovery_failed"` records failures. When comparing cluster load, include lock acquisition/refresh RPCs and the leader's cleanup backlog as well as namespace traversal work. ## Metadata key conventions