fix(ecstore): remove reachable panics in tiering, replication, and heal paths (#4205)

* fix(ecstore): remove reachable panics in tiering, replication, and heal paths

- Parse x-amz-expiration leniently in tier PUT responses; any lifecycle
  rule on the remote tier bucket returns an RFC1123 date that the previous
  ISO8601 unwrap turned into a panic of the ILM transition worker
- Skip invalid user-metadata header values (with a warning) when building
  tier and replication PUT headers instead of panicking on non-ASCII input
- Heal: tolerate absent data_dir for delete markers and remote objects
- transition_object: don't unwrap version_id on unversioned buckets when
  recording partial writes for offline disks
- Admin server info: use port_or_known_default() so default-port (80/443)
  endpoints don't panic is_server_resolvable
- Tier ListObjectsV2 client: decode response body with from_utf8_lossy
- walk_internal: log merge_entry_channels errors instead of dropping them

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(ecstore): fail heal explicitly when data_dir is missing

Address review feedback: unwrap_or_default() silently substituted a nil
UUID when latest metadata lacked data_dir. Delete markers and remote
objects legitimately have no data_dir and skip the data-heal block, but
for a regular object a missing data_dir means corrupt metadata — return
FileCorrupt with a descriptive log instead of building part paths under
a nil UUID directory.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -1254,6 +1254,17 @@ impl PutObjectOptions {
         }
     }
 
+    /// Insert `value` as header `name`, skipping values that are not valid
+    /// HTTP header values (with a warning) instead of panicking mid-replication.
+    fn insert_checked(header: &mut HeaderMap, name: &'static str, value: &str) {
+        match HeaderValue::from_str(value) {
+            Ok(v) => {
+                header.insert(name, v);
+            }
+            Err(_) => warn!("skipping header {} with invalid value", name),
+        }
+    }
+
     pub fn header(&self) -> HeaderMap {
         let mut header = HeaderMap::new();
 
@@ -1261,66 +1272,75 @@ impl PutObjectOptions {
         if content_type.is_empty() {
             content_type = "application/octet-stream".to_string();
         }
-        header.insert("Content-Type", HeaderValue::from_str(&content_type).expect("err"));
+        match HeaderValue::from_str(&content_type) {
+            Ok(v) => {
+                header.insert("Content-Type", v);
+            }
+            Err(_) => {
+                warn!("invalid Content-Type header value, falling back to application/octet-stream");
+                header.insert("Content-Type", HeaderValue::from_static("application/octet-stream"));
+            }
+        }
 
         if !self.content_encoding.is_empty() {
-            header.insert("Content-Encoding", HeaderValue::from_str(&self.content_encoding).expect("err"));
+            Self::insert_checked(&mut header, "Content-Encoding", &self.content_encoding);
         }
         if !self.content_disposition.is_empty() {
-            header.insert("Content-Disposition", HeaderValue::from_str(&self.content_disposition).expect("err"));
+            Self::insert_checked(&mut header, "Content-Disposition", &self.content_disposition);
         }
         if !self.content_language.is_empty() {
-            header.insert("Content-Language", HeaderValue::from_str(&self.content_language).expect("err"));
+            Self::insert_checked(&mut header, "Content-Language", &self.content_language);
         }
         if !self.cache_control.is_empty() {
-            header.insert("Cache-Control", HeaderValue::from_str(&self.cache_control).expect("err"));
+            Self::insert_checked(&mut header, "Cache-Control", &self.cache_control);
         }
 
         if self.expires.unix_timestamp() != 0 {
-            header.insert("Expires", HeaderValue::from_str(&self.expires.format(&Rfc3339).unwrap()).expect("err")); //rustfs invalid header
+            match self.expires.format(&Rfc3339) {
+                Ok(expires) => Self::insert_checked(&mut header, "Expires", &expires),
+                Err(err) => warn!("skipping Expires header, format failed: {}", err),
+            }
         }
 
         if let Some(mode) = &self.mode {
-            header.insert(AMZ_OBJECT_LOCK_MODE, HeaderValue::from_str(mode.as_str()).expect("err"));
+            Self::insert_checked(&mut header, AMZ_OBJECT_LOCK_MODE, mode.as_str());
         }
 
         if self.retain_until_date.unix_timestamp() != 0 {
-            header.insert(
+            match self.retain_until_date.format(&Rfc3339) {
-                AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
+                Ok(retain_until) => Self::insert_checked(&mut header, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, &retain_until),
-                HeaderValue::from_str(&self.retain_until_date.format(&Rfc3339).unwrap()).expect("err"),
+                Err(err) => warn!("skipping object-lock retain-until-date header, format failed: {}", err),
-            );
+            }
         }
 
         if let Some(legalhold) = &self.legalhold {
-            header.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD, HeaderValue::from_str(legalhold.as_str()).expect("err"));
+            Self::insert_checked(&mut header, AMZ_OBJECT_LOCK_LEGAL_HOLD, legalhold.as_str());
         }
 
         if !self.storage_class.is_empty() {
-            header.insert(AMZ_STORAGE_CLASS, HeaderValue::from_str(&self.storage_class).expect("err"));
+            Self::insert_checked(&mut header, AMZ_STORAGE_CLASS, &self.storage_class);
         }
 
         if !self.website_redirect_location.is_empty() {
-            header.insert(
+            Self::insert_checked(&mut header, AMZ_WEBSITE_REDIRECT_LOCATION, &self.website_redirect_location);
-                AMZ_WEBSITE_REDIRECT_LOCATION,
-                HeaderValue::from_str(&self.website_redirect_location).expect("err"),
-            );
         }
 
         if !self.internal.replication_status.as_str().is_empty() {
-            header.insert(
+            Self::insert_checked(&mut header, AMZ_BUCKET_REPLICATION_STATUS, self.internal.replication_status.as_str());
-                AMZ_BUCKET_REPLICATION_STATUS,
-                HeaderValue::from_str(self.internal.replication_status.as_str()).expect("err"),
-            );
         }
 
         for (k, v) in &self.user_metadata {
+            let Ok(header_value) = HeaderValue::from_str(v) else {
+                warn!("skipping user metadata header with invalid value: {}", k);
+                continue;
+            };
             if is_amz_header(k) || is_standard_header(k) || is_storageclass_header(k) || is_rustfs_header(k) || is_minio_header(k)
             {
                 if let Ok(header_name) = HeaderName::from_bytes(k.as_bytes()) {
-                    header.insert(header_name, HeaderValue::from_str(v).unwrap());
+                    header.insert(header_name, header_value);
                 }
             } else if let Ok(header_name) = HeaderName::from_bytes(format!("x-amz-meta-{k}").as_bytes()) {
-                header.insert(header_name, HeaderValue::from_str(v).unwrap());
+                header.insert(header_name, header_value);
             }
         }
 

crates/ecstore/src/client/api_list.rs

@@ -133,7 +133,7 @@ impl TransitionClient {
                 body_vec.extend_from_slice(data);
             }
         }
-        let mut list_bucket_result = match quick_xml::de::from_str::<ListBucketV2Result>(&String::from_utf8(body_vec).unwrap()) {
+        let mut list_bucket_result = match quick_xml::de::from_str::<ListBucketV2Result>(&String::from_utf8_lossy(&body_vec)) {
             Ok(result) => result,
             Err(err) => {
                 return Err(std::io::Error::other(err.to_string()));

crates/ecstore/src/client/api_put_object.rs

@@ -234,13 +234,17 @@ impl PutObjectOptions {
         }
 
         for (k, v) in &self.user_metadata {
+            let Ok(header_value) = HeaderValue::from_str(v) else {
+                warn!("skipping user metadata header with invalid value: {}", k);
+                continue;
+            };
             if is_amz_header(k) || is_standard_header(k) || is_storageclass_header(k) || is_rustfs_header(k) || is_minio_header(k)
             {
                 if let Ok(header_name) = HeaderName::from_bytes(k.as_bytes()) {
-                    header.insert(header_name, HeaderValue::from_str(&v).expect("operation should succeed"));
+                    header.insert(header_name, header_value);
                 }
             } else if let Ok(header_name) = HeaderName::from_bytes(format!("x-amz-meta-{}", k).as_bytes()) {
-                header.insert(header_name, HeaderValue::from_str(&v).expect("operation should succeed"));
+                header.insert(header_name, header_value);
             }
         }
 

crates/ecstore/src/client/api_put_object_multipart.rs

@@ -388,14 +388,13 @@ impl TransitionClient {
 
         let complete_multipart_upload_result: CompleteMultipartUploadResult = CompleteMultipartUploadResult::default();
 
-        let (exp_time, rule_id) = if let Some(h_x_amz_expiration) = resp.headers().get(X_AMZ_EXPIRATION) {
+        let exp_time = resp
-            (
+            .headers()
-                OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(),
+            .get(X_AMZ_EXPIRATION)
-                "".to_string(),
+            .and_then(|v| v.to_str().ok())
-            )
+            .and_then(|s| OffsetDateTime::parse(s, ISO8601_DATEFORMAT).ok())
-        } else {
+            .unwrap_or_else(OffsetDateTime::now_utc);
-            (OffsetDateTime::now_utc(), "".to_string())
+        let rule_id = "".to_string();
-        };
 
         Ok(UploadInfo {
             bucket: complete_multipart_upload_result.bucket,

crates/ecstore/src/client/api_put_object_streaming.rs

@@ -523,14 +523,13 @@ impl TransitionClient {
             )));
         }
 
-        let (exp_time, rule_id) = if let Some(h_x_amz_expiration) = resp.headers().get(X_AMZ_EXPIRATION) {
+        let exp_time = resp
-            (
+            .headers()
-                OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(),
+            .get(X_AMZ_EXPIRATION)
-                "".to_string(),
+            .and_then(|v| v.to_str().ok())
-            )
+            .and_then(|s| OffsetDateTime::parse(s, ISO8601_DATEFORMAT).ok())
-        } else {
+            .unwrap_or_else(OffsetDateTime::now_utc);
-            (OffsetDateTime::now_utc(), "".to_string())
+        let rule_id = "".to_string();
-        };
         let h = resp.headers();
         Ok(UploadInfo {
             bucket: bucket_name.to_string(),

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -79,7 +79,9 @@ async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> {
         "{}://{}:{}",
         endpoint.url.scheme(),
         endpoint.url.host_str().expect("URL should have host"),
-        endpoint.url.port().expect("URL should have port")
+        // `Url::port()` is None when the URL uses the scheme's default port
+        // (e.g. http on 80 / https on 443); fall back to the scheme default.
+        endpoint.url.port_or_known_default().expect("URL should have port")
     );
 
     let ping_task = async {

crates/ecstore/src/set_disk/heal.rs

@@ -352,8 +352,26 @@ impl SetDisks {
 
                         // We write at temporary location and then rename to final location.
                         let tmp_id = Uuid::new_v4().to_string();
-                        let src_data_dir = latest_meta.data_dir.expect("operation should succeed").to_string();
+                        // Delete markers and remote (transitioned) objects carry no data_dir and
-                        let dst_data_dir = latest_meta.data_dir.expect("operation should succeed");
+                        // skip the data-heal block below, so a nil placeholder is safe for them.
+                        // For a regular object a missing data_dir means the latest metadata is
+                        // corrupt; fail this object's heal with a clear error instead of building
+                        // part paths under a nil UUID directory.
+                        let data_dir = match latest_meta.data_dir {
+                            Some(data_dir) => data_dir,
+                            None => {
+                                if !latest_meta.deleted && !latest_meta.is_remote() {
+                                    error!(
+                                        "heal: latest metadata for {}/{} has no data_dir, cannot heal object data",
+                                        bucket, object
+                                    );
+                                    return Err(DiskError::FileCorrupt);
+                                }
+                                Uuid::nil()
+                            }
+                        };
+                        let src_data_dir = data_dir.to_string();
+                        let dst_data_dir = data_dir;
 
                         if !latest_meta.deleted && !latest_meta.is_remote() {
                             let erasure_info = latest_meta.erasure.clone();

crates/ecstore/src/set_disk/mod.rs

@@ -4446,7 +4446,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             if let Some(disk) = disk {
                 continue;
             }
-            let _ = self.add_partial(bucket, object, opts.version_id.as_ref().expect("err")).await;
+            let _ = self
+                .add_partial(bucket, object, opts.version_id.as_deref().unwrap_or_default())
+                .await;
             break;
         }
 

crates/ecstore/src/store/list_objects.rs

@@ -1883,7 +1883,14 @@ impl ECStore {
             .instrument(tracing::Span::current()),
         );
 
-        tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await }.instrument(tracing::Span::current()));
+        tokio::spawn(
+            async move {
+                if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
+                    error!("merge_entry_channels err {:?}", err)
+                }
+            }
+            .instrument(tracing::Span::current()),
+        );
 
         let walk_started = std::time::Instant::now();
         let walk_results = join_all(futures).await;
@@ -2952,7 +2959,14 @@ impl Sets {
             .instrument(tracing::Span::current()),
         );
 
-        tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await }.instrument(tracing::Span::current()));
+        tokio::spawn(
+            async move {
+                if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
+                    error!("merge_entry_channels err {:?}", err)
+                }
+            }
+            .instrument(tracing::Span::current()),
+        );
 
         let walk_started = std::time::Instant::now();
         let walk_results = join_all(futures).await;