diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index f0d26f72a..ec0b3c692 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1249,13 +1249,12 @@ impl TransitionState { return false; } let bucket = bucket.to_string(); - let scheduled = Arc::clone(&self.compensation_buckets); let state = Arc::clone(self); tokio::spawn(async move { Self::inc_counter(&state.compensation_running_tasks); state.record_scanner_transition_state(); let Some(api) = runtime_sources::object_store_handle() else { - scheduled.lock().unwrap().remove(&bucket); + state.finish_bucket_compensation(&bucket); Self::add_counter(&state.compensation_running_tasks, -1); state.record_scanner_transition_state(); debug!( @@ -1291,13 +1290,25 @@ impl TransitionState { ); } - scheduled.lock().unwrap().remove(&bucket); + state.finish_bucket_compensation(&bucket); Self::add_counter(&state.compensation_running_tasks, -1); state.record_scanner_transition_state(); }); true } + fn finish_bucket_compensation(&self, bucket: &str) { + match self.compensation_buckets.lock() { + Ok(mut scheduled) => { + scheduled.remove(bucket); + } + Err(poisoned) => { + poisoned.into_inner().remove(bucket); + self.compensation_buckets.clear_poison(); + } + } + } + #[inline] fn inc_counter(counter: &AtomicI64) { Self::add_counter(counter, 1); @@ -1729,7 +1740,7 @@ impl TransitionState { } pub fn add_lastday_stats(&self, tier: &str, ts: TierStats) { - let mut tier_stats = self.last_day_stats.lock().unwrap(); + let mut tier_stats = self.lock_last_day_stats(); tier_stats .entry(tier.to_string()) .and_modify(|e| e.add_stats(ts)) @@ -1737,7 +1748,7 @@ impl TransitionState { } pub fn get_daily_all_tier_stats(&self) -> DailyAllTierStats { - let tier_stats = self.last_day_stats.lock().unwrap(); + let tier_stats = self.lock_last_day_stats(); let mut res = DailyAllTierStats::with_capacity(tier_stats.len()); for (tier, st) in tier_stats.iter() { res.insert(tier.clone(), st.clone()); @@ -1745,6 +1756,18 @@ impl TransitionState { res } + fn lock_last_day_stats(&self) -> std::sync::MutexGuard<'_, HashMap> { + match self.last_day_stats.lock() { + Ok(stats) => stats, + Err(poisoned) => { + let mut stats = poisoned.into_inner(); + stats.clear(); + self.last_day_stats.clear_poison(); + stats + } + } + } + pub async fn update_workers(api: Arc, n: i64) { Self::update_workers_inner(api, n).await; } @@ -1766,7 +1789,27 @@ impl TransitionState { fn resize_workers_to(api: Arc, n: i64, requested: i64, absolute_max: i64) { let target = n as usize; let transition_state = runtime_sources::transition_state_handle(); - let mut workers = transition_state.workers.lock().unwrap(); + let runtime = match tokio::runtime::Handle::try_current() { + Ok(runtime) => runtime, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + error = %err, + state = "resize_failed", + "Lifecycle worker pool requires a Tokio runtime" + ); + return; + } + }; + // Runtime lookup happens before locking, and the guard is dropped before + // metrics/logging callbacks. Poison therefore means a Vec mutation may + // have unwound and worker tracking cannot be reconstructed safely. + let mut workers = transition_state + .workers + .lock() + .expect("transition worker tracking mutex poisoned"); let tracked_workers = workers.len(); workers.retain(|worker| !worker.handle.is_finished()); let pruned_finished_workers = tracked_workers.saturating_sub(workers.len()); @@ -1776,7 +1819,7 @@ impl TransitionState { let clone_api = api.clone(); let cancel = CancellationToken::new(); let worker_cancel = cancel.clone(); - let handle = tokio::spawn(async move { + let handle = runtime.spawn(async move { TransitionState::worker_with_cancel(clone_api, worker_cancel).await; }); workers.push(TransitionWorker { cancel, handle }); @@ -1790,6 +1833,7 @@ impl TransitionState { let current_workers = workers.len() as i64; transition_state.num_workers.store(current_workers, Ordering::SeqCst); + drop(workers); transition_state.record_scanner_transition_state(); debug!( @@ -4884,6 +4928,7 @@ mod tests { FreeVersionRecoveryStats, RecoveryWalkTestAction, list_tier_free_versions, recover_tier_free_versions_with_cancel, set_recovery_bucket_list_wait_hook, set_recovery_walk_test_hook, }; + use crate::bucket::lifecycle::tier_last_day_stats::LastDayTierStats; use crate::bucket::lifecycle::tier_sweeper::Jentry; use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; use crate::bucket::metadata_sys; @@ -4917,6 +4962,7 @@ mod tests { use http::HeaderMap; use rustfs_common::metrics::{IlmAction, global_metrics}; use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX; + use rustfs_data_usage::TierStats; use rustfs_filemeta::{FileInfo, FileMeta}; use s3s::dto::{ BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry, OutputLocation, @@ -6984,6 +7030,46 @@ mod tests { assert_eq!(state.compensation_pending_tasks(), 1); } + #[test] + fn poisoned_compensation_set_can_release_completed_bucket() { + let state = TransitionState::new_with_capacity(1); + state + .compensation_buckets + .lock() + .expect("fresh mutex should lock") + .insert("bucket-a".to_string()); + let poison_target = Arc::clone(&state.compensation_buckets); + let _ = std::thread::spawn(move || { + let _guard = poison_target.lock().expect("fresh mutex should lock"); + panic!("poison compensation set"); + }) + .join(); + + state.finish_bucket_compensation("bucket-a"); + assert_eq!(state.compensation_pending_tasks(), 0); + assert!( + state.compensation_buckets.lock().is_ok(), + "validated compensation state must clear poison" + ); + } + + #[test] + fn poisoned_tier_stats_are_reset_before_reuse() { + let state = TransitionState::new_with_capacity(1); + let poison_target = Arc::clone(&state.last_day_stats); + let _ = std::thread::spawn(move || { + let mut stats = poison_target.lock().expect("fresh mutex should lock"); + stats.insert("stale".to_string(), LastDayTierStats::default()); + panic!("poison tier stats"); + }) + .join(); + + state.add_lastday_stats("fresh", TierStats::default()); + let stats = state.get_daily_all_tier_stats(); + assert!(!stats.contains_key("stale"), "possibly partial statistics must be discarded"); + assert!(stats.contains_key("fresh"), "statistics must accept new samples after recovery"); + } + #[tokio::test(flavor = "current_thread")] async fn scanner_transition_state_reports_compensation_pending_buckets() { let state = TransitionState::new_with_capacity(1); @@ -7274,6 +7360,23 @@ mod tests { TransitionState::resize_workers_to(ecstore, original_workers, original_workers, absolute_max); } + #[tokio::test] + #[serial] + async fn transition_worker_resize_without_runtime_does_not_poison_tracking() { + let (_paths, ecstore) = setup_test_env().await; + let transition_state = runtime_sources::transition_state_handle(); + let resize = std::thread::spawn(move || { + TransitionState::resize_workers_to(ecstore, 1, 1, resolve_transition_workers_absolute_max()); + }) + .join(); + + assert!(resize.is_ok(), "missing Tokio runtime must not panic while worker tracking is locked"); + assert!( + transition_state.workers.lock().is_ok(), + "failed resize must leave worker tracking unpoisoned" + ); + } + #[test] fn should_defer_date_expiry_for_recent_config_update_respects_grace_window() { let now = OffsetDateTime::now_utc(); diff --git a/crates/ecstore/src/client/api_put_object_streaming.rs b/crates/ecstore/src/client/api_put_object_streaming.rs index fdb1e820e..2df913a06 100644 --- a/crates/ecstore/src/client/api_put_object_streaming.rs +++ b/crates/ecstore/src/client/api_put_object_streaming.rs @@ -22,7 +22,7 @@ use bytes::Bytes; use futures::future::join_all; use http::{HeaderMap, HeaderName, HeaderValue, StatusCode}; use std::io::Error; -use std::sync::RwLock; +use std::sync::{Mutex, MutexGuard, RwLock}; use std::{collections::HashMap, sync::Arc}; use time::{OffsetDateTime, format_description}; use tokio::io::AsyncReadExt; @@ -46,6 +46,14 @@ use crate::client::utils::base64_encode; use rustfs_utils::path::trim_etag; use s3s::header::X_AMZ_EXPIRATION; +fn lock_md5_hasher( + md5_hasher: &Mutex>, +) -> Result>, std::io::Error> { + md5_hasher + .lock() + .map_err(|_| std::io::Error::other("MD5 hasher state is unavailable")) +} + /// Read exactly `want` bytes for a single multipart part, or fewer if the reader /// reaches EOF first. Advances the reader so the next call returns the following /// part. Replaces the previous per-part `read_all()`/`to_vec()`, which drained @@ -177,7 +185,7 @@ impl TransitionClient { let length = buf.len(); if opts.send_content_md5 { - let mut md5_hasher = self.md5_hasher.lock().unwrap(); + let mut md5_hasher = lock_md5_hasher(&self.md5_hasher)?; let md5_hash = match md5_hasher.as_mut() { Some(hasher) => hasher, None => return Err(std::io::Error::other("MD5 hasher not initialized")), @@ -370,7 +378,7 @@ impl TransitionClient { let mut md5_base64: String = "".to_string(); if opts.send_content_md5 { - let mut md5_hasher = clone_self.md5_hasher.lock().unwrap(); + let mut md5_hasher = lock_md5_hasher(&clone_self.md5_hasher)?; let md5_hash = match md5_hasher.as_mut() { Some(hasher) => hasher, None => { @@ -418,6 +426,9 @@ impl TransitionClient { } let results = join_all(futures).await; + for result in results { + result?; + } select! { err = err_rx.recv() => { @@ -620,10 +631,12 @@ fn collect_complete_parts(parts_info: &HashMap, total_parts_cou #[cfg(test)] mod tests { - use super::{ObjectPart, ReaderImpl, collect_complete_parts, read_multipart_part}; + use super::{ObjectPart, ReaderImpl, collect_complete_parts, lock_md5_hasher, read_multipart_part}; use crate::object_api::GetObjectReader; use bytes::Bytes; + use rustfs_utils::hash::HashAlgorithm; use std::collections::HashMap; + use std::sync::{Arc, Mutex}; // Drive a reader through the same per-part loop the multipart stream uses and // collect the size of every part. Regression for rustfs/rustfs#4811: the old @@ -733,4 +746,18 @@ mod tests { "a gap in the parts map must be an error, not a panic" ); } + + #[test] + fn poisoned_md5_state_fails_closed() { + let hasher = Arc::new(Mutex::new(Some(HashAlgorithm::Md5))); + let poison_target = Arc::clone(&hasher); + let _ = std::thread::spawn(move || { + let _guard = poison_target.lock().expect("fresh mutex should lock"); + panic!("poison MD5 state"); + }) + .join(); + + let error = lock_md5_hasher(&hasher).expect_err("poisoned hash state must not be reused"); + assert_eq!(error.kind(), std::io::ErrorKind::Other); + } } diff --git a/crates/ecstore/src/set_disk/ops/heal_walk.rs b/crates/ecstore/src/set_disk/ops/heal_walk.rs index 70291b039..3a87a0898 100644 --- a/crates/ecstore/src/set_disk/ops/heal_walk.rs +++ b/crates/ecstore/src/set_disk/ops/heal_walk.rs @@ -69,6 +69,13 @@ struct HealWalkCollector { } impl HealWalkCollector { + fn lock_objects(&self) -> disk::error::Result>> { + self.objects.lock().map_err(|_| { + self.cancel.cancel(); + DiskError::FileCorrupt + }) + } + /// Expand one resolved entry into its versions and record it. Cancels the /// walk once EITHER page bound (distinct object names OR expanded versions) /// is met — always at a sorted object-key boundary so a heavily-versioned @@ -105,7 +112,9 @@ impl HealWalkCollector { let added = versions.len(); let (objs_len, ver_total) = { - let mut objects = self.objects.lock().unwrap(); + let Ok(mut objects) = self.lock_objects() else { + return; + }; objects.push(HealWalkObject { name: entry.name, versions, @@ -239,7 +248,10 @@ impl SetDisks { Err(err) => return Err(err), } - let objects = std::mem::take(&mut *collector.objects.lock().unwrap()); + let objects = { + let mut objects = collector.lock_objects()?; + std::mem::take(&mut *objects) + }; let truncated = collector.truncated.load(Ordering::SeqCst); Ok(finalize_heal_walk_page(objects, forward_to, truncated)) } @@ -310,4 +322,28 @@ mod tests { assert_eq!(next_forward, None, "a complete final page must not carry a resume cursor"); assert!(!truncated); } + + #[test] + fn poisoned_collector_state_cancels_the_walk() { + let collector = Arc::new(HealWalkCollector { + bucket: "bucket".to_string(), + batch_objects: 2, + version_budget: 2, + objects: Mutex::new(Vec::new()), + version_total: AtomicUsize::new(0), + truncated: AtomicBool::new(false), + cancel: CancellationToken::new(), + }); + let poison_target = Arc::clone(&collector); + let _ = std::thread::spawn(move || { + let _guard = poison_target.objects.lock().expect("fresh mutex should lock"); + panic!("poison heal collector"); + }) + .join(); + + let error = collector.lock_objects().expect_err("poisoned collection must fail closed"); + assert_eq!(error, DiskError::FileCorrupt); + assert!(collector.cancel.is_cancelled(), "a poisoned page collector must cancel its walk"); + assert!(collector.objects.lock().is_err(), "poisoned state must remain fail-closed"); + } }