From dd6b5525e4bccb3bfefeac0e150c9ff6bcffeaf7 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 3 Jul 2026 00:14:11 +0800 Subject: [PATCH] fix(ecstore): remove reachable panics in tiering, replication, and heal paths (#4205) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 --------- Co-authored-by: Claude Fable 5 --- .../ecstore/src/bucket/bucket_target_sys.rs | 66 ++++++++++++------- crates/ecstore/src/client/api_list.rs | 2 +- crates/ecstore/src/client/api_put_object.rs | 8 ++- .../src/client/api_put_object_multipart.rs | 15 ++--- .../src/client/api_put_object_streaming.rs | 15 ++--- .../src/diagnostics/admin_server_info.rs | 4 +- crates/ecstore/src/set_disk/heal.rs | 22 ++++++- crates/ecstore/src/set_disk/mod.rs | 4 +- crates/ecstore/src/store/list_objects.rs | 18 ++++- 9 files changed, 106 insertions(+), 48 deletions(-) diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index f068b8b3d..70190cda0 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/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( - AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, - HeaderValue::from_str(&self.retain_until_date.format(&Rfc3339).unwrap()).expect("err"), - ); + match self.retain_until_date.format(&Rfc3339) { + Ok(retain_until) => Self::insert_checked(&mut header, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, &retain_until), + 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( - AMZ_WEBSITE_REDIRECT_LOCATION, - HeaderValue::from_str(&self.website_redirect_location).expect("err"), - ); + Self::insert_checked(&mut header, AMZ_WEBSITE_REDIRECT_LOCATION, &self.website_redirect_location); } if !self.internal.replication_status.as_str().is_empty() { - header.insert( - AMZ_BUCKET_REPLICATION_STATUS, - HeaderValue::from_str(self.internal.replication_status.as_str()).expect("err"), - ); + Self::insert_checked(&mut header, AMZ_BUCKET_REPLICATION_STATUS, self.internal.replication_status.as_str()); } 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); } } diff --git a/crates/ecstore/src/client/api_list.rs b/crates/ecstore/src/client/api_list.rs index 061ce48ed..b1eaf0e56 100644 --- a/crates/ecstore/src/client/api_list.rs +++ b/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::(&String::from_utf8(body_vec).unwrap()) { + let mut list_bucket_result = match quick_xml::de::from_str::(&String::from_utf8_lossy(&body_vec)) { Ok(result) => result, Err(err) => { return Err(std::io::Error::other(err.to_string())); diff --git a/crates/ecstore/src/client/api_put_object.rs b/crates/ecstore/src/client/api_put_object.rs index aa068e0c4..4d661bee9 100644 --- a/crates/ecstore/src/client/api_put_object.rs +++ b/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); } } diff --git a/crates/ecstore/src/client/api_put_object_multipart.rs b/crates/ecstore/src/client/api_put_object_multipart.rs index 29e245ad7..66c621016 100644 --- a/crates/ecstore/src/client/api_put_object_multipart.rs +++ b/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) { - ( - OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(), - "".to_string(), - ) - } else { - (OffsetDateTime::now_utc(), "".to_string()) - }; + let exp_time = resp + .headers() + .get(X_AMZ_EXPIRATION) + .and_then(|v| v.to_str().ok()) + .and_then(|s| OffsetDateTime::parse(s, ISO8601_DATEFORMAT).ok()) + .unwrap_or_else(OffsetDateTime::now_utc); + let rule_id = "".to_string(); Ok(UploadInfo { bucket: complete_multipart_upload_result.bucket, diff --git a/crates/ecstore/src/client/api_put_object_streaming.rs b/crates/ecstore/src/client/api_put_object_streaming.rs index 3e95eeeda..9fdcb9b8a 100644 --- a/crates/ecstore/src/client/api_put_object_streaming.rs +++ b/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) { - ( - OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(), - "".to_string(), - ) - } else { - (OffsetDateTime::now_utc(), "".to_string()) - }; + let exp_time = resp + .headers() + .get(X_AMZ_EXPIRATION) + .and_then(|v| v.to_str().ok()) + .and_then(|s| OffsetDateTime::parse(s, ISO8601_DATEFORMAT).ok()) + .unwrap_or_else(OffsetDateTime::now_utc); + let rule_id = "".to_string(); let h = resp.headers(); Ok(UploadInfo { bucket: bucket_name.to_string(), diff --git a/crates/ecstore/src/diagnostics/admin_server_info.rs b/crates/ecstore/src/diagnostics/admin_server_info.rs index cb48b09e7..50d9a4231 100644 --- a/crates/ecstore/src/diagnostics/admin_server_info.rs +++ b/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 { diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index 3b71027d6..f242463e5 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/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(); - let dst_data_dir = latest_meta.data_dir.expect("operation should succeed"); + // Delete markers and remote (transitioned) objects carry no data_dir and + // 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(); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 7df65ac60..5d68ccf76 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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; } diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 179a843ee..dae93b2cb 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/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;