fix(ecstore): reject over-NAME_MAX key segments up front; classify irreconcilable parity as corrupt metadata (#5804)

fix(ecstore): reject over-NAME_MAX object key segments up front and classify irreconcilable parity as corrupt metadata

Two defects found during release acceptance and the backlog#1776 investigation:

Object keys with any path segment longer than 255 bytes could never be stored (each segment maps to one on-disk directory entry), but the failure surfaced only when the disk layer hit ENAMETOOLONG, which leaked to clients as InternalError 500 (rustfs#5785). Validate the on-disk segment budget in check_bucket_and_object_names so such keys fail deterministically as ObjectNameInvalid (4xx) before any I/O. Directory-object keys (trailing '/') account for the __XLDIR__ suffix their final segment carries on disk.

object_quorum_from_meta conflated two very different no-quorum situations (rustfs#5801): stray or foreign metadata whose parity values are garbage produced the same retryable-looking ErasureReadQuorum (503) as a genuine partial outage, so clients retried unrecoverable reads and monitoring could not tell corruption from capacity loss. Now (a) parity counts outside [0, total_shards] are treated as invalid entries instead of being clamped to i32::MAX, which could poison common_parity's occurrence counting, and (b) when a full read quorum of disks answers but their parity values cannot be reconciled, the error is FileCorrupt — heal-actionable and non-retryable — while too-few-healthy-replies keeps returning ErasureReadQuorum.

Verification: 4 new unit tests (segment budget boundaries incl. byte-vs-char and __XLDIR__ budget; garbage parity sanitization; corrupt-vs-quorum classification), metadata::tests + utils::tests 62/62, set_disk+bucket suites 1214 passed with the single pre-existing heal_queue_marks_missing_versioning_state_as_missed cross-test flake also failing on a clean tree (not introduced here), clippy clean, make pre-commit green.

crates/ecstore/src/bucket/utils.rs

@@ -269,6 +269,32 @@ pub fn check_del_obj_args(bucket: &str, object: &str) -> Result<()> {
     check_bucket_and_object_names(bucket, object)
 }
 
+/// Filesystem `NAME_MAX`: every object-key path segment becomes one on-disk
+/// directory entry, so a longer segment can never be stored and previously
+/// escaped as an `ENAMETOOLONG` io error → `InternalError` 500 (rustfs#5785).
+const MAX_OBJECT_KEY_SEGMENT_BYTES: usize = 255;
+
+/// Reject object keys whose on-disk directory names would exceed `NAME_MAX`.
+///
+/// Middle segments map to their raw bytes; the final segment of a
+/// directory-object key (trailing `/`) is stored with the `__XLDIR__` suffix
+/// appended, shrinking its budget accordingly.
+fn object_key_segments_fit_on_disk(object: &str) -> bool {
+    let trailing_dir = object.ends_with('/');
+    let segments: Vec<&str> = object.split('/').collect();
+    let last_nonempty = segments.iter().rposition(|s| !s.is_empty());
+    for (index, segment) in segments.iter().enumerate() {
+        let mut budget = MAX_OBJECT_KEY_SEGMENT_BYTES;
+        if trailing_dir && Some(index) == last_nonempty {
+            budget = budget.saturating_sub(rustfs_utils::path::GLOBAL_DIR_SUFFIX.len());
+        }
+        if segment.len() > budget {
+            return false;
+        }
+    }
+    true
+}
+
 pub fn check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
     if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() {
         return Err(StorageError::BucketNameInvalid(bucket.to_string()));
@@ -282,6 +308,10 @@ pub fn check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
         return Err(StorageError::ObjectNameInvalid(bucket.to_string(), object.to_string()));
     }
 
+    if !object_key_segments_fit_on_disk(object) {
+        return Err(StorageError::ObjectNameInvalid(bucket.to_string(), object.to_string()));
+    }
+
     // if cfg!(target_os = "windows") && object.contains('\\') {
     //     return Err(StorageError::ObjectNameInvalid(bucket.to_string(), object.to_string()));
     // }
@@ -387,6 +417,37 @@ mod tests {
     use super::*;
     use proptest::prelude::*;
 
+    /// rustfs#5785: keys whose path segments exceed the on-disk NAME_MAX
+    /// budget must be rejected up front as ObjectNameInvalid (4xx), not leak
+    /// ENAMETOOLONG as InternalError 500 from the disk layer.
+    #[test]
+    fn object_key_segment_name_max_budget() {
+        // 255-byte single segment: exactly at the on-disk limit.
+        assert!(check_bucket_and_object_names("bucket", &"a".repeat(255)).is_ok());
+        // 256 bytes: one over.
+        assert!(matches!(
+            check_bucket_and_object_names("bucket", &"a".repeat(256)),
+            Err(StorageError::ObjectNameInvalid(_, _))
+        ));
+        // Long keys are fine as long as every segment fits.
+        let segmented = ["b".repeat(200), "c".repeat(200), "d".repeat(200)].join("/");
+        assert!(check_bucket_and_object_names("bucket", &segmented).is_ok());
+        // The budget counts bytes, not characters (100 CJK chars = 300 bytes).
+        assert!(matches!(
+            check_bucket_and_object_names("bucket", &"中".repeat(100)),
+            Err(StorageError::ObjectNameInvalid(_, _))
+        ));
+        assert!(check_bucket_and_object_names("bucket", &"中".repeat(85)).is_ok());
+        // Directory-object keys spend GLOBAL_DIR_SUFFIX bytes of the final
+        // segment's budget on the on-disk __XLDIR__ encoding.
+        let dir_budget = 255 - rustfs_utils::path::GLOBAL_DIR_SUFFIX.len();
+        assert!(check_bucket_and_object_names("bucket", &format!("{}/", "e".repeat(dir_budget))).is_ok());
+        assert!(matches!(
+            check_bucket_and_object_names("bucket", &format!("{}/", "e".repeat(dir_budget + 1))),
+            Err(StorageError::ObjectNameInvalid(_, _))
+        ));
+    }
+
     // Test validation functions
     #[test]
     fn test_is_valid_object_name() {

crates/ecstore/src/set_disk/metadata.rs

@@ -250,14 +250,26 @@ impl SetDisks {
                 continue;
             }
 
+            // A parity count outside [0, total_shards] cannot describe a real
+            // layout on this set: it comes from corrupt or foreign metadata
+            // (e.g. stray leftovers, rustfs#5801). Treat the entry as invalid
+            // instead of clamping to i32::MAX, which would poison
+            // `common_parity`'s occurrence counting.
+            let erasure_parity = i32::try_from(metadata.erasure.parity_blocks).unwrap_or(-1);
+            let erasure_parity = if (0..=total_shards_i32).contains(&erasure_parity) {
+                erasure_parity
+            } else {
+                -1
+            };
             if metadata.is_canonical_delete_marker() || metadata.size == 0 {
                 parities[index] = half;
+            } else if erasure_parity < 0 {
+                parities[index] = -1;
             } else if metadata.transition_status == TRANSITION_COMPLETE {
                 let majority_metadata_parity = total_shards_i32 - (half + 1);
-                let erasure_parity = i32::try_from(metadata.erasure.parity_blocks).unwrap_or(i32::MAX);
                 parities[index] = majority_metadata_parity.max(erasure_parity);
             } else {
-                parities[index] = i32::try_from(metadata.erasure.parity_blocks).unwrap_or(i32::MAX);
+                parities[index] = erasure_parity;
             }
         }
         parities
@@ -294,6 +306,19 @@ impl SetDisks {
         let parity_blocks = Self::common_parity(&parities, default_parity_count as i32);
 
         if parity_blocks < 0 {
+            // No parity value reached read quorum. Distinguish two cases:
+            // enough disks answered with valid-looking metadata that simply
+            // cannot be reconciled (corrupt/foreign entries — retrying cannot
+            // help, and heal should see Corrupt, rustfs#5801) versus too few
+            // healthy answers (a genuine quorum condition where retry may
+            // succeed once disks recover).
+            let healthy_replies = errs.iter().filter(|err| err.is_none()).count();
+            if healthy_replies >= expected_rquorum {
+                error!(
+                    "object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}"
+                );
+                return Err(DiskError::FileCorrupt);
+            }
             error!("object_quorum_from_meta: parity_blocks < 0, errs={:?}", errs);
             return Err(DiskError::ErasureReadQuorum);
         }
@@ -1136,7 +1161,9 @@ mod tests {
         let invalid = vec![FileInfo::default(); 4];
         let err = SetDisks::object_quorum_from_meta(&invalid, &vec![None; 4], 2)
             .expect_err("invalid metadata without a common parity must fail closed");
-        assert_eq!(err, DiskError::ErasureReadQuorum);
+        // A full set of healthy replies whose metadata cannot be reconciled is
+        // corrupt (heal-actionable), not a retryable quorum outage (rustfs#5801).
+        assert_eq!(err, DiskError::FileCorrupt);
     }
 
     #[test]
@@ -1468,4 +1495,55 @@ mod tests {
             "compatible prefixes carrying the same mapping must share one identity"
         );
     }
+
+    /// rustfs#5801: parity counts outside [0, total_shards] come from corrupt
+    /// or foreign metadata and must be treated as invalid entries instead of
+    /// clamped values that poison `common_parity`'s occurrence counting.
+    #[test]
+    fn out_of_range_parity_is_treated_as_invalid_entry() {
+        let mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let mut metas: Vec<FileInfo> = (0..4).map(|i| metadata_quorum_test_fileinfo(mod_time, i)).collect();
+        for fi in &mut metas {
+            fi.erasure.parity_blocks = usize::MAX;
+        }
+        let errs: Vec<Option<DiskError>> = vec![None; 4];
+
+        let parities = SetDisks::list_object_parities(&metas, &errs);
+        assert_eq!(parities, vec![-1; 4], "garbage parity must not survive as a candidate");
+    }
+
+    /// rustfs#5801: when a read quorum of healthy disks answers but their
+    /// parity values are irreconcilable, the object metadata is corrupt —
+    /// return `FileCorrupt` (heal-actionable, non-retryable) instead of the
+    /// retryable-looking `ErasureReadQuorum`.
+    #[test]
+    fn irreconcilable_parity_with_healthy_quorum_is_file_corrupt() {
+        let mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let mut metas: Vec<FileInfo> = (0..4).map(|i| metadata_quorum_test_fileinfo(mod_time, i)).collect();
+        for fi in &mut metas {
+            fi.erasure.parity_blocks = usize::MAX;
+        }
+        let errs: Vec<Option<DiskError>> = vec![None; 4];
+
+        let err = SetDisks::object_quorum_from_meta(&metas, &errs, 2).expect_err("garbage parity cannot form a quorum");
+        assert_eq!(err, DiskError::FileCorrupt);
+    }
+
+    /// Too few healthy replies remains a genuine quorum condition where a
+    /// retry may succeed once disks recover.
+    #[test]
+    fn insufficient_healthy_replies_stays_erasure_read_quorum() {
+        let mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let mut metas: Vec<FileInfo> = (0..4).map(|i| metadata_quorum_test_fileinfo(mod_time, i)).collect();
+        metas[0].erasure.parity_blocks = usize::MAX;
+        let errs: Vec<Option<DiskError>> = vec![
+            None,
+            Some(DiskError::DiskNotFound),
+            Some(DiskError::DiskNotFound),
+            Some(DiskError::DiskNotFound),
+        ];
+
+        let err = SetDisks::object_quorum_from_meta(&metas, &errs, 2).expect_err("one healthy reply is below quorum");
+        assert_eq!(err, DiskError::ErasureReadQuorum);
+    }
 }