fix(filemeta): validate part array lengths in into_fileinfo (#4382)

MetaObject::into_fileinfo indexed part_sizes[i]/part_actual_sizes[i] by
part_numbers.len() without checking the arrays are the same length,
unlike the adjacent part_etags/part_indices which are length-guarded.
decode_from pushes the three arrays independently and the xl.meta CRC
only covers bytes, so a CRC-valid but internally inconsistent xl.meta
(foreign writer / MinIO interop) triggers an out-of-bounds panic on the
GET/HEAD/LIST decode path.

Guard the three arrays for equal length and return Err(FileCorrupt) so a
divergent shard is skipped and quorum uses the other disks, instead of
panicking the request task. Cascade into_fileinfo to Result across its
callers, and fix io_primitives early-return to derive the version id from
the merged header and fall into the per-disk loop (single-disk survival +
heal). The 2118 merge-first path is left as a documented follow-up.

Refs backlog#900 (filemeta-01).

crates/filemeta/src/filemeta.rs

@@ -716,9 +716,21 @@ impl FileMeta {
                     && let Ok(found_free_fi) = ver.parse_version_meta()
                     && found_free_fi.version_type != VersionType::Invalid
                 {
-                    let mut free_fi = found_free_fi.into_fileinfo(volume, path, all_parts);
-                    free_fi.is_latest = true;
-                    found_free_version = Some(free_fi);
+                    // Graceful degradation: the free-version replication-accounting record
+                    // is auxiliary metadata; a corrupt one must not tank an otherwise
+                    // healthy primary-version read. Log and skip rather than propagate.
+                    // Known side effect: if a disk holds only free versions and they are
+                    // corrupt, `into_fileinfo` falls through to `FileNotFound` (not
+                    // `FileCorrupt`), so that disk is not enqueued for heal.
+                    match found_free_fi.into_fileinfo(volume, path, all_parts) {
+                        Ok(mut free_fi) => {
+                            free_fi.is_latest = true;
+                            found_free_version = Some(free_fi);
+                        }
+                        Err(e) => {
+                            warn!(volume, path, error = %e, "skipping corrupt free version during into_fileinfo");
+                        }
+                    }
                 }
 
                 if header.version_id != Some(vid) {
@@ -1543,6 +1555,84 @@ mod test {
         }
     }
 
+    // ------------------------------------------------------------------
+    // backlog#900: CRC-valid but semantically corrupt part arrays must
+    // produce Err(FileCorrupt), never panic.
+    // ------------------------------------------------------------------
+
+    fn valid_object_version(version_id: Uuid, part_sizes: Vec<usize>) -> FileMetaVersion {
+        FileMetaVersion {
+            version_type: VersionType::Object,
+            object: Some(MetaObject {
+                version_id: Some(version_id),
+                erasure_algorithm: ErasureAlgo::ReedSolomon,
+                erasure_m: 2,
+                erasure_n: 2,
+                erasure_block_size: 1 << 20,
+                bitrot_checksum_algo: ChecksumAlgo::HighwayHash,
+                part_numbers: vec![1, 2],
+                part_sizes, // caller-injected (short = corrupt)
+                part_actual_sizes: vec![10, 20],
+                mod_time: Some(OffsetDateTime::now_utc()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn crc_valid_but_part_arrays_corrupt_into_fileinfo_errors_not_panics() {
+        // A short part_sizes Object version round-trips through the real codec: the CRC is
+        // valid and load succeeds, but into_fileinfo(all_parts) hits the length guard and
+        // returns Err(FileCorrupt) rather than panicking.
+        let mut fm = FileMeta::new();
+        fm.add_version_filemata(valid_object_version(Uuid::new_v4(), vec![10]))
+            .expect("add corrupt-parts version");
+
+        let encoded = fm.marshal_msg().expect("marshal recomputes a valid CRC");
+        let loaded = FileMeta::load(&encoded).expect("CRC-valid meta must load (lazy, parts not decoded yet)");
+
+        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            loaded.into_fileinfo("bucket", "key", "", true, false, true)
+        }));
+        let inner = caught.expect("into_fileinfo must not panic on CRC-valid but semantically corrupt parts");
+        assert!(matches!(inner, Err(Error::FileCorrupt)), "expected FileCorrupt");
+    }
+
+    proptest! {
+        #[test]
+        fn into_fileinfo_never_panics_on_arbitrary_loaded_meta(input in vec(any::<u8>(), 0..=4096)) {
+            // Complements filemeta_load_never_panics_on_arbitrary_bytes: for every FileMeta
+            // that loads, into_fileinfo(all_parts=true) must not panic (Ok or Err).
+            if let Ok(fm) = FileMeta::load(&input) {
+                let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                    fm.into_fileinfo("b", "k", "", true, false, true)
+                }));
+                prop_assert!(caught.is_ok(), "into_fileinfo panicked on loaded meta");
+            }
+        }
+    }
+
+    #[test]
+    fn into_file_info_versions_fails_whole_listing_on_one_corrupt_version() {
+        // One corrupt version among healthy ones fails the whole listing (into_file_info_versions
+        // uses `?`, so no partial results). This is the established failure semantics of the
+        // merge-first exact-versions path (backlog#900 §3.3), strictly better than a panic.
+        let healthy_id = Uuid::new_v4();
+        let corrupt_id = Uuid::new_v4();
+        let mut fm = FileMeta::new();
+        fm.add_version_filemata(valid_object_version(healthy_id, vec![10, 20]))
+            .expect("healthy");
+        fm.add_version_filemata(valid_object_version(corrupt_id, vec![10]))
+            .expect("corrupt"); // short part_sizes
+
+        let fm = FileMeta::load(&fm.marshal_msg().expect("marshal")).expect("load");
+
+        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| fm.into_file_info_versions("bucket", "key", true)));
+        let inner = caught.expect("must not panic");
+        assert!(matches!(inner, Err(Error::FileCorrupt)), "whole-listing must fail with FileCorrupt");
+    }
+
     #[test]
     fn test_performance_with_large_metadata() {
         // Test performance with large metadata files
@@ -2055,7 +2145,7 @@ mod test {
         fm.update_object_version(update).unwrap();
 
         let (_, version) = fm.find_version(version_id).unwrap();
-        let stored = version.into_fileinfo("bucket", "test", true);
+        let stored = version.into_fileinfo("bucket", "test", true).expect("into_fileinfo");
         assert_eq!(stored.metadata.get("x-amz-meta-owner"), Some(&"alice".to_string()));
         assert_eq!(stored.checksum, Some(checksum));
     }