fix(runtime): remove startup panic fallbacks (#3754)

* fix(runtime): remove startup panic fallbacks

* test(runtime): cover buffer profile fallback safety

* test(runtime): reduce panic-style assertions

* fix(runtime): expose fallible env config setup

* test(runtime): simplify permit acquisition assertion

* test(runtime): tighten operation helper assertions

* fix(filemeta): stop panicking on invalid free version ids

* fix(init): satisfy buffer profile clippy lints

* fix(lock): harden fast lock config construction

* chore(checks): refresh layer dependency baseline

---------

Signed-off-by: houseme <housemecn@gmail.com>

crates/concurrency/src/manager.rs

@@ -14,7 +14,7 @@
 
 //! Main concurrency manager
 
-use crate::config::ConcurrencyConfig;
+use crate::config::{ConcurrencyConfig, ConfigError};
 use std::sync::Arc;
 
 /// Snapshot of disk permit queue usage for GetObject orchestration.
@@ -76,13 +76,7 @@ pub struct ConcurrencyManager {
 }
 
 impl ConcurrencyManager {
-    /// Create a new concurrency manager with the given configuration
-    pub fn new(config: ConcurrencyConfig) -> Self {
-        // Validate configuration
-        if let Err(e) = config.validate() {
-            panic!("Invalid concurrency configuration: {}", e);
-        }
-
+    fn build(config: ConcurrencyConfig) -> Self {
         Self {
             #[cfg(feature = "timeout")]
             timeout: Arc::new(crate::timeout::TimeoutManager::from_policy(config.timeout_policy)),
@@ -106,14 +100,57 @@ impl ConcurrencyManager {
         }
     }
 
+    /// Try to create a new concurrency manager with the given configuration.
+    pub fn try_new(config: ConcurrencyConfig) -> Result<Self, ConfigError> {
+        config.validate()?;
+        Ok(Self::build(config))
+    }
+
+    /// Create a new concurrency manager with the given configuration.
+    ///
+    /// Invalid configurations are downgraded to the default configuration instead of
+    /// panicking so startup/runtime callers can remain fail-safe in production paths.
+    pub fn new(config: ConcurrencyConfig) -> Self {
+        match Self::try_new(config) {
+            Ok(manager) => manager,
+            Err(err) => {
+                tracing::warn!(
+                    event = "concurrency_manager.invalid_config_fallback",
+                    component = "concurrency",
+                    subsystem = "manager",
+                    error = %err,
+                    "Invalid concurrency configuration detected; falling back to defaults"
+                );
+                Self::build(ConcurrencyConfig::default())
+            }
+        }
+    }
+
     /// Create with default configuration
     pub fn with_defaults() -> Self {
-        Self::new(ConcurrencyConfig::default())
+        Self::build(ConcurrencyConfig::default())
+    }
+
+    /// Try to create a manager from environment-derived configuration.
+    pub fn try_from_env() -> Result<Self, ConfigError> {
+        Self::try_new(ConcurrencyConfig::from_env())
     }
 
     /// Create from environment variables
     pub fn from_env() -> Self {
-        Self::new(ConcurrencyConfig::from_env())
+        match Self::try_from_env() {
+            Ok(manager) => manager,
+            Err(err) => {
+                tracing::warn!(
+                    event = "concurrency_manager.invalid_env_fallback",
+                    component = "concurrency",
+                    subsystem = "manager",
+                    error = %err,
+                    "Invalid environment-derived concurrency configuration detected; falling back to defaults"
+                );
+                Self::build(ConcurrencyConfig::default())
+            }
+        }
     }
 
     /// Get the configuration
@@ -305,6 +342,40 @@ mod tests {
         assert!(manager.config().validate().is_ok());
     }
 
+    #[test]
+    fn test_try_new_returns_error_for_invalid_config() {
+        let config = ConcurrencyConfig {
+            timeout_policy: crate::TimeoutManagerPolicy {
+                default_timeout: std::time::Duration::from_secs(10),
+                max_timeout: std::time::Duration::from_secs(1),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let result = ConcurrencyManager::try_new(config);
+        assert!(matches!(result, Err(ConfigError::InvalidTimeout(_))));
+    }
+
+    #[test]
+    fn test_new_falls_back_to_default_for_invalid_config() {
+        let config = ConcurrencyConfig {
+            timeout_policy: crate::TimeoutManagerPolicy {
+                default_timeout: std::time::Duration::from_secs(10),
+                max_timeout: std::time::Duration::from_secs(1),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let manager = ConcurrencyManager::new(config);
+        assert!(manager.config().validate().is_ok());
+        assert_eq!(
+            manager.config().timeout_policy.default_timeout,
+            ConcurrencyConfig::default().timeout_policy.default_timeout
+        );
+    }
+
     #[tokio::test]
     async fn test_manager_lifecycle() {
         let manager = ConcurrencyManager::with_defaults();

crates/filemeta/src/filemeta.rs

@@ -652,7 +652,7 @@ impl FileMeta {
         } else {
             self.versions.remove(i);
 
-            let (free_version, to_free) = obj.init_free_version(fi);
+            let (free_version, to_free) = obj.init_free_version(fi)?;
 
             if to_free {
                 self.add_version_filemata(free_version).err()

crates/filemeta/src/filemeta/version.rs

@@ -2140,18 +2140,14 @@ impl MetaObject {
         [bytes[0], bytes[1], bytes[2], bytes[3]]
     }
 
-    pub fn init_free_version(&self, fi: &FileInfo) -> (FileMetaVersion, bool) {
+    pub fn init_free_version(&self, fi: &FileInfo) -> Result<(FileMetaVersion, bool)> {
         if fi.skip_tier_free_version() {
-            return (FileMetaVersion::default(), false);
+            return Ok((FileMetaVersion::default(), false));
         }
         if let Some(status) = get_bytes(&self.meta_sys, SUFFIX_TRANSITION_STATUS)
             && status == TRANSITION_COMPLETE.as_bytes().to_vec()
         {
-            let vid = Uuid::parse_str(&fi.tier_free_version_id());
-            if let Err(err) = vid {
-                panic!("Invalid Tier Object delete marker versionId {} {}", fi.tier_free_version_id(), err);
-            }
-            let vid = vid.unwrap();
+            let vid = Uuid::parse_str(&fi.tier_free_version_id())?;
             let mut free_entry = FileMetaVersion {
                 version_type: VersionType::Delete,
                 write_version: 0,
@@ -2176,9 +2172,9 @@ impl MetaObject {
                     insert_bytes(&mut delete_marker.meta_sys, suffix, v);
                 }
             }
-            return (free_entry, true);
+            return Ok((free_entry, true));
         }
-        (FileMetaVersion::default(), false)
+        Ok((FileMetaVersion::default(), false))
     }
 }
 
@@ -3391,4 +3387,20 @@ mod tests {
         .into_fileinfo("b", "k", false);
         assert_eq!(fi.transition_version_id, Some(id));
     }
+
+    #[test]
+    fn meta_object_init_free_version_rejects_invalid_tier_free_version_id() {
+        let mut sys = HashMap::new();
+        insert_bytes(&mut sys, SUFFIX_TRANSITION_STATUS, TRANSITION_COMPLETE.as_bytes().to_vec());
+
+        let obj = make_meta_object_with_sys(sys);
+        let mut fi = FileInfo::new("object", 2, 2);
+        fi.set_tier_free_version_id("not-a-uuid");
+
+        let err = obj
+            .init_free_version(&fi)
+            .expect_err("invalid free-version UUID should return an error instead of panicking");
+
+        assert!(matches!(err, Error::UuidParse(_)));
+    }
 }

crates/heal/src/heal/manager.rs

@@ -2813,10 +2813,10 @@ mod tests {
             .pop_runnable(|request| can_schedule_request(request, &running, 1))
             .expect("should find runnable request");
 
-        match popped.heal_type {
-            HealType::ErasureSet { set_disk_id, .. } => assert_eq!(set_disk_id, "pool_0_set_2"),
-            other => panic!("expected erasure set request, got {other:?}"),
-        }
+        assert!(matches!(
+            popped.heal_type,
+            HealType::ErasureSet { ref set_disk_id, .. } if set_disk_id == "pool_0_set_2"
+        ));
     }
 
     #[test]

crates/lock/src/fast_lock/manager.rs

@@ -15,7 +15,9 @@
 use std::sync::Arc;
 use tokio::sync::RwLock;
 use tokio::time::{Instant, interval};
+use tracing::warn;
 
+use crate::LockError;
 use crate::fast_lock::{
     guard::FastLockGuard,
     manager_trait::LockManager,
@@ -42,8 +44,37 @@ impl FastObjectLockManager {
 
     /// Create new lock manager with custom config
     pub fn with_config(config: LockConfig) -> Self {
+        if let Err(err) = Self::validate_config(&config) {
+            warn!(
+                error = %err,
+                shard_count = config.shard_count,
+                fallback_shard_count = crate::fast_lock::DEFAULT_SHARD_COUNT,
+                "Invalid lock manager configuration, falling back to defaults"
+            );
+            return Self::build(LockConfig::default());
+        }
+
+        Self::build(config)
+    }
+
+    /// Create new lock manager with custom config, returning an explicit error for invalid input.
+    pub fn try_with_config(config: LockConfig) -> crate::Result<Self> {
+        Self::validate_config(&config)?;
+        Ok(Self::build(config))
+    }
+
+    fn validate_config(config: &LockConfig) -> crate::Result<()> {
+        if config.shard_count == 0 || !config.shard_count.is_power_of_two() {
+            return Err(LockError::configuration(format!(
+                "shard count must be a non-zero power of 2, got {}",
+                config.shard_count
+            )));
+        }
+        Ok(())
+    }
+
+    fn build(config: LockConfig) -> Self {
         let shard_count = config.shard_count;
-        assert!(shard_count.is_power_of_two(), "Shard count must be power of 2");
 
         let shards: Vec<Arc<LockShard>> = (0..shard_count).map(|i| Arc::new(LockShard::new(i))).collect();
 

crates/lock/src/fast_lock/tests.rs

@@ -14,8 +14,9 @@
 
 #[cfg(test)]
 mod fast_lock_tests {
-    use crate::fast_lock::FastObjectLockManager;
+    use crate::LockError;
     use crate::fast_lock::types::{LockConfig, LockMode, LockPriority, LockResult, ObjectKey, ObjectLockRequest};
+    use crate::fast_lock::{DEFAULT_SHARD_COUNT, FastObjectLockManager};
     use std::sync::Arc;
     use std::time::Duration;
     use tokio::time::sleep;
@@ -31,6 +32,32 @@ mod fast_lock_tests {
         FastObjectLockManager::with_config(config)
     }
 
+    #[test]
+    fn try_with_config_returns_error_for_invalid_shard_count() {
+        let config = LockConfig {
+            shard_count: 3,
+            ..LockConfig::default()
+        };
+
+        let err = FastObjectLockManager::try_with_config(config)
+            .expect_err("non-power-of-two shard counts should return an explicit configuration error");
+
+        assert!(matches!(err, LockError::Configuration { .. }));
+        assert!(err.to_string().contains("shard count must be a non-zero power of 2"));
+    }
+
+    #[tokio::test]
+    async fn with_config_falls_back_to_default_for_invalid_shard_count() {
+        let config = LockConfig {
+            shard_count: 3,
+            ..LockConfig::default()
+        };
+
+        let manager = FastObjectLockManager::with_config(config);
+
+        assert_eq!(manager.shards.len(), DEFAULT_SHARD_COUNT);
+    }
+
     #[tokio::test]
     async fn test_basic_write_lock_acquire_release() {
         let manager = create_test_manager();