fix(head): clearer NoSuchKey for prefix keys and validate part numbers (#1638)

rustfs/src/storage/ecfs.rs

@@ -19,7 +19,7 @@ use crate::server::RemoteAddr;
 use crate::storage::concurrency::{
     CachedGetObject, ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager,
 };
-use crate::storage::entity;
+use crate::storage::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children};
 use crate::storage::helper::OperationHelper;
 use crate::storage::options::{filter_object_metadata, get_content_sha256};
 use crate::storage::{
@@ -36,6 +36,7 @@ use crate::storage::{
     process_topic_configurations, strip_managed_encryption_metadata, validate_bucket_object_lock_enabled,
     validate_list_object_unordered_with_delimiter, validate_object_key, wrap_response_with_cors,
 };
+use crate::storage::{entity, parse_part_number_i32_to_usize};
 use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
 use bytes::Bytes;
 use datafusion::arrow::{
@@ -3394,14 +3395,8 @@ impl S3 for FS {
 
         // Validate object key
         validate_object_key(&key, "HEAD")?;
-
+        // Parse part number from Option<i32> to Option<usize> with validation
-        let part_number = part_number.map(|v| v as usize);
+        let part_number: Option<usize> = parse_part_number_i32_to_usize(part_number, "HEAD")?;
-
-        if let Some(part_num) = part_number
-            && part_num == 0
-        {
-            return Err(s3_error!(InvalidArgument, "part_number invalid"));
-        }
 
         let rs = range.map(|v| match v {
             Range::Int { first, last } => HTTPRangeSpec {
@@ -3427,27 +3422,35 @@ impl S3 for FS {
         let Some(store) = new_object_layer_fn() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
-
         // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors
         let info = match store.get_object_info(&bucket, &key, &opts).await {
             Ok(info) => info,
             Err(err) => {
                 // If the error indicates the object or its version was not found, return 404 (NoSuchKey)
                 if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
+                    if is_dir_object(&key) {
+                        let has_children = match probe_prefix_has_children(store, &bucket, &key, false).await {
+                            Ok(has_children) => has_children,
+                            Err(e) => {
+                                error!("Failed to probe children for prefix (bucket: {}, key: {}): {}", bucket, key, e);
+                                false
+                            }
+                        };
+                        let msg = head_prefix_not_found_message(&bucket, &key, has_children);
+                        return Err(S3Error::with_message(S3ErrorCode::NoSuchKey, msg));
+                    }
                     return Err(S3Error::new(S3ErrorCode::NoSuchKey));
                 }
                 // Other errors, such as insufficient permissions, still return the original error
                 return Err(ApiError::from(err).into());
             }
         };
-
         if info.delete_marker {
             if opts.version_id.is_none() {
                 return Err(S3Error::new(S3ErrorCode::NoSuchKey));
             }
             return Err(S3Error::new(S3ErrorCode::MethodNotAllowed));
         }
-
         if let Some(match_etag) = if_none_match
             && let Some(strong_etag) = match_etag.into_etag()
             && info
@@ -3457,7 +3460,6 @@ impl S3 for FS {
         {
             return Err(S3Error::new(S3ErrorCode::NotModified));
         }
-
         if let Some(modified_since) = if_modified_since {
             // obj_time < givenTime + 1s
             if info.mod_time.is_some_and(|mod_time| {
@@ -3467,7 +3469,6 @@ impl S3 for FS {
                 return Err(S3Error::new(S3ErrorCode::NotModified));
             }
         }
-
         if let Some(match_etag) = if_match {
             if let Some(strong_etag) = match_etag.into_etag()
                 && info
@@ -3485,7 +3486,6 @@ impl S3 for FS {
         {
             return Err(S3Error::new(S3ErrorCode::PreconditionFailed));
         }
-
         let event_info = info.clone();
         let content_type = {
             if let Some(content_type) = &info.content_type {
@@ -3526,7 +3526,6 @@ impl S3 for FS {
             .or_else(|| metadata_map.get("x-amz-storage-class").cloned())
             .filter(|s| !s.is_empty())
             .map(StorageClass::from);
-
         let mut checksum_crc32 = None;
         let mut checksum_crc32c = None;
         let mut checksum_sha1 = None;
@@ -3543,7 +3542,6 @@ impl S3 for FS {
                 .decrypt_checksums(opts.part_number.unwrap_or(0), &req.headers)
                 .map_err(ApiError::from)?;
 
-            debug!("get object metadata checksums: {:?}", checksums);
             for (key, checksum) in checksums {
                 if key == AMZ_CHECKSUM_TYPE {
                     checksum_type = Some(ChecksumType::from(checksum));
@@ -3560,7 +3558,6 @@ impl S3 for FS {
                 }
             }
         }
-
         // Extract standard HTTP headers from user_defined metadata
         // Note: These headers are stored with lowercase keys by extract_metadata_from_mime
         let cache_control = metadata_map.get("cache-control").cloned();
@@ -3576,7 +3573,6 @@ impl S3 for FS {
         } else {
             0
         };
-
         let output = HeadObjectOutput {
             content_length: Some(content_length),
             content_type,
@@ -3649,10 +3645,8 @@ impl S3 for FS {
         {
             response.headers.insert(header_name, header_value);
         }
-
         let result = Ok(response);
         let _ = helper.complete(&result);
-
         result
     }
 

rustfs/src/storage/ecfs_extend.rs

@@ -1102,3 +1102,27 @@ pub(crate) async fn wrap_response_with_cors<T>(
 
     response
 }
+
+/// Parse part number from Option<i32> to Option<usize> with validation
+/// This function checks that the part number is greater than 0 and
+/// converts it to usize, returning an error if invalid
+///
+/// # Arguments
+/// * `part_number` - The optional part number as i32
+/// * `op` - The operation name for logging purposes
+///
+/// # Returns
+/// * `Ok(Some(usize))` if part number is valid
+/// * `Ok(None)` if part number is None
+/// * `Err(S3Error)` if part number is invalid (0 or overflow)
+#[inline]
+pub(crate) fn parse_part_number_i32_to_usize(part_number: Option<i32>, op: &'static str) -> S3Result<Option<usize>> {
+    match part_number {
+        None => Ok(None),
+        Some(n) if n <= 0 => Err(S3Error::with_message(
+            S3ErrorCode::InvalidArgument,
+            format!("{op}: invalid partNumber {n}, must be a positive integer"),
+        )),
+        Some(n) => Ok(Some(n as usize)),
+    }
+}

rustfs/src/storage/head_prefix.rs

@@ -0,0 +1,75 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use rustfs_ecstore::StorageAPI;
+use rustfs_ecstore::store::ECStore;
+use std::sync::Arc;
+
+/// Determines if the key "looks like a prefix" (ends with `/`).
+/// Note: No special handling for empty strings here; the caller must ensure the key has passed `validate_object_key`.
+#[allow(dead_code)]
+#[inline]
+pub(crate) fn is_prefix_key(key: &str) -> bool {
+    key.ends_with('/')
+}
+
+/// Constructs a more explicit error message when `HEAD` is performed on a `prefix`-style key but the directory marker object is missing.
+///
+/// `has_children`:
+/// - true: Indicates that there are objects under the prefix, but the "directory marker object" does not exist (common in semantics relying solely on prefix listing).
+/// - false: Indicates that there are no objects under the prefix either.
+pub(crate) fn head_prefix_not_found_message(bucket: &str, key: &str, has_children: bool) -> String {
+    if has_children {
+        format!(
+            "NoSuchKey: key `{key}` looks like a prefix (ends with `/`), prefix has children objects, \
+but directory marker object does not exist in bucket `{bucket}`"
+        )
+    } else {
+        format!(
+            "NoSuchKey: key `{key}` looks like a prefix (ends with `/`), but no directory marker object \
+and no objects exist under this prefix in bucket `{bucket}`"
+        )
+    }
+}
+
+/// Lightweight probe: Checks if any objects exist under the prefix (max_keys=1).
+///
+/// Purpose: Only used to optimize the error message, without changing error codes or external semantics.
+pub(crate) async fn probe_prefix_has_children(
+    store: Arc<ECStore>,
+    bucket: &str,
+    prefix: &str,
+    include_deleted: bool,
+) -> Result<bool, String> {
+    // Even if the underlying implementation has overhead for list, this is only one extra request, and it only happens when NotFound occurs and the key ends with `/`.
+    // No delimiter is introduced here to avoid "virtual directory" hierarchy affecting the judgment: as long as there are any objects under the prefix.
+    let res = store
+        .list_objects_v2(
+            bucket,
+            prefix,
+            None,  // continuation_token
+            None,  // delimiter
+            1,     // max_keys
+            false, // fetch_owner
+            None,  // start_after
+            include_deleted,
+        )
+        .await
+        .map_err(|e| format!("{e}"))?;
+
+    let has_objects = !res.objects.is_empty();
+    let has_common_prefixes = !res.prefixes.is_empty();
+
+    Ok(has_objects || has_common_prefixes)
+}

rustfs/src/storage/mod.rs

@@ -25,5 +25,6 @@ mod concurrent_get_object_test;
 mod ecfs_extend;
 #[cfg(test)]
 mod ecfs_test;
+pub(crate) mod head_prefix;
 
 pub(crate) use ecfs_extend::*;