fix(ecstore): handle mutex poisoning explicitly (#5379)

* fix(ecstore): handle mutex poisoning explicitly

* test(ecstore): satisfy poison assertion lint

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<String, LastDayTierStats>> {
+        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<ECStore>, n: i64) {
         Self::update_workers_inner(api, n).await;
     }
@@ -1766,7 +1789,27 @@ impl TransitionState {
     fn resize_workers_to(api: Arc<ECStore>, 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();

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<Option<rustfs_utils::hash::HashAlgorithm>>,
+) -> Result<MutexGuard<'_, Option<rustfs_utils::hash::HashAlgorithm>>, 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<i64, ObjectPart>, 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);
+    }
 }

crates/ecstore/src/set_disk/ops/heal_walk.rs

@@ -69,6 +69,13 @@ struct HealWalkCollector {
 }
 
 impl HealWalkCollector {
+    fn lock_objects(&self) -> disk::error::Result<std::sync::MutexGuard<'_, Vec<HealWalkObject>>> {
+        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");
+    }
 }