fix(ecstore): require commit quorum for latest metadata (#4117)

crates/filemeta/src/filemeta/version.rs

@@ -271,6 +271,18 @@ pub struct FileMetaShallowVersion {
     pub meta: Vec<u8>, // FileMetaVersion.marshal_msg
 }
 
+fn write_quorum_from_erasure(data_blocks: usize, parity_blocks: usize) -> Option<usize> {
+    if data_blocks == 0 {
+        return None;
+    }
+
+    Some(if data_blocks == parity_blocks {
+        data_blocks.saturating_add(1)
+    } else {
+        data_blocks
+    })
+}
+
 impl FileMetaShallowVersion {
     /// Parse version meta with legacy format compatibility.
     /// Use this instead of `FileMetaVersion::default()` + `unmarshal_msg()` to handle old-version xl.meta.
@@ -278,6 +290,18 @@ impl FileMetaShallowVersion {
         FileMetaVersion::try_from(self.meta.as_slice())
     }
 
+    pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
+        let header_quorum = self.header.write_quorum(fallback_quorum);
+        if self.header.has_ec() || !matches!(self.header.version_type, VersionType::Object | VersionType::Legacy) {
+            return header_quorum;
+        }
+
+        self.parse_version_meta()
+            .ok()
+            .and_then(|version| version.erasure_write_quorum())
+            .unwrap_or(header_quorum)
+    }
+
     pub fn into_fileinfo(&self, volume: &str, path: &str, all_parts: bool) -> Result<FileInfo> {
         let file_version = self.parse_version_meta()?;
         Ok(file_version.into_fileinfo(volume, path, all_parts))
@@ -312,6 +336,23 @@ pub struct FileMetaVersion {
 }
 
 impl FileMetaVersion {
+    fn erasure_write_quorum(&self) -> Option<usize> {
+        let (data_blocks, parity_blocks) = match self.version_type {
+            VersionType::Object | VersionType::Legacy => {
+                if let Some(object) = &self.object {
+                    (object.erasure_m, object.erasure_n)
+                } else if let Some(object) = &self.legacy_object {
+                    (object.erasure.data_blocks, object.erasure.parity_blocks)
+                } else {
+                    return None;
+                }
+            }
+            _ => return None,
+        };
+
+        write_quorum_from_erasure(data_blocks, parity_blocks)
+    }
+
     fn decode_data_dir_from_v2_object(buf: &[u8]) -> Result<Option<Uuid>> {
         let mut cur = std::io::Cursor::new(buf);
         let mut fields = rmp::decode::read_map_len(&mut cur)?;
@@ -775,6 +816,14 @@ impl FileMetaVersionHeader {
         self.ec_m > 0 && self.ec_n > 0
     }
 
+    pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
+        if self.version_type != VersionType::Object || !self.has_ec() {
+            return fallback_quorum;
+        }
+
+        write_quorum_from_erasure(usize::from(self.ec_m), usize::from(self.ec_n)).unwrap_or(fallback_quorum)
+    }
+
     pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool {
         let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o);
         if self.version_id.is_none() {
@@ -2543,9 +2592,28 @@ pub enum Flags {
 
 // mergeXLV2Versions
 pub fn merge_file_meta_versions(
+    quorum: usize,
+    strict: bool,
+    requested_versions: usize,
+    versions: &[Vec<FileMetaShallowVersion>],
+) -> Vec<FileMetaShallowVersion> {
+    merge_file_meta_versions_inner(quorum, strict, requested_versions, false, versions)
+}
+
+pub(crate) fn merge_file_meta_versions_with_write_quorum(
+    quorum: usize,
+    strict: bool,
+    requested_versions: usize,
+    versions: &[Vec<FileMetaShallowVersion>],
+) -> Vec<FileMetaShallowVersion> {
+    merge_file_meta_versions_inner(quorum, strict, requested_versions, true, versions)
+}
+
+fn merge_file_meta_versions_inner(
     mut quorum: usize,
     mut strict: bool,
     requested_versions: usize,
+    enforce_write_quorum: bool,
     versions: &[Vec<FileMetaShallowVersion>],
 ) -> Vec<FileMetaShallowVersion> {
     if quorum == 0 {
@@ -2557,13 +2625,32 @@ pub fn merge_file_meta_versions(
     }
 
     if versions.len() == 1 {
-        return versions[0].clone();
+        if !enforce_write_quorum {
+            return versions[0].clone();
+        }
+
+        let required_quorum = versions[0]
+            .first()
+            .map(|version| version.write_quorum(quorum).max(quorum))
+            .unwrap_or(quorum);
+        if versions.len() >= required_quorum {
+            return versions[0].clone();
+        }
+        return Vec::new();
     }
 
     if quorum == 1 {
         strict = true;
     }
 
+    let required_quorum = |version: &FileMetaShallowVersion| {
+        if enforce_write_quorum {
+            version.write_quorum(quorum).max(quorum)
+        } else {
+            quorum
+        }
+    };
+
     let mut versions = versions.to_owned();
 
     let mut n_versions = 0;
@@ -2594,9 +2681,12 @@ pub fn merge_file_meta_versions(
 
         let mut latest = FileMetaShallowVersion::default();
         if consistent {
-            merged.push(tops[0].clone());
-            if !tops[0].header.free_version() {
-                n_versions += 1;
+            latest = tops[0].clone();
+            if tops.len() >= required_quorum(&latest) {
+                merged.push(latest.clone());
+                if !latest.header.free_version() {
+                    n_versions += 1;
+                }
             }
         } else {
             let mut latest_count = 0;
@@ -2659,7 +2749,7 @@ pub fn merge_file_meta_versions(
                     break;
                 }
             }
-            if latest_count >= quorum {
+            if latest_count >= required_quorum(&latest) {
                 if !latest.header.free_version() {
                     n_versions += 1;
                 }
@@ -3053,6 +3143,39 @@ mod tests {
         wr
     }
 
+    fn encode_legacy_v2_object_body(data_blocks: usize, parity_blocks: usize) -> Vec<u8> {
+        let drive_count = data_blocks + parity_blocks;
+        let payload = LegacyObjectVersionFixture {
+            version_type: LegacyObjectVersionTypeFixture::Object,
+            object: Some(LegacyObjectFixture {
+                version_id: Some(sample_version_id().as_bytes().to_vec()),
+                data_dir: Some(Uuid::from_u128(42).as_bytes().to_vec()),
+                erasure_algorithm: "ReedSolomon".to_string(),
+                erasure_m: data_blocks,
+                erasure_n: parity_blocks,
+                erasure_block_size: 1_048_576,
+                erasure_index: 1,
+                erasure_dist: (1..=drive_count)
+                    .map(|idx| u8::try_from(idx).expect("test drive index should fit u8"))
+                    .collect(),
+                bitrot_checksum_algo: "HighwayHash".to_string(),
+                part_numbers: vec![1],
+                part_etags: vec!["etag-1".to_string()],
+                part_sizes: vec![11],
+                part_actual_sizes: vec![11],
+                part_indices: vec![Vec::new()],
+                size: 11,
+                mod_time: Some(sample_mod_time()),
+                meta_sys: HashMap::new(),
+                meta_user: HashMap::from([("content-type".to_string(), "text/plain".to_string())]),
+            }),
+            delete_marker: None,
+            write_version: 3,
+        };
+
+        rmp_serde::to_vec_named(&payload).expect("legacy object payload should marshal")
+    }
+
     #[test]
     fn version_header_unmarshal_v1_uses_legacy_layout_defaults() {
         let expected = sample_header();
@@ -3087,6 +3210,38 @@ mod tests {
         assert_eq!(decoded.ec_m, 0);
     }
 
+    #[test]
+    fn shallow_version_write_quorum_uses_legacy_object_payload_when_header_lacks_ec() {
+        let expected = sample_header();
+        let encoded = encode_v2_header(&expected);
+        let mut header = FileMetaVersionHeader::default();
+        header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
+
+        let version = FileMetaShallowVersion {
+            header,
+            meta: encode_legacy_v2_object_body(7, 1),
+        };
+
+        assert_eq!(version.header.write_quorum(5), 5);
+        assert_eq!(version.write_quorum(5), 7);
+    }
+
+    #[test]
+    fn shallow_version_write_quorum_uses_zero_parity_object_payload() {
+        let expected = sample_header();
+        let encoded = encode_v2_header(&expected);
+        let mut header = FileMetaVersionHeader::default();
+        header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
+
+        let version = FileMetaShallowVersion {
+            header,
+            meta: encode_legacy_v2_object_body(4, 0),
+        };
+
+        assert_eq!(version.header.write_quorum(2), 2);
+        assert_eq!(version.write_quorum(2), 4);
+    }
+
     #[test]
     fn version_header_unmarshal_v3_round_trips_current_layout() {
         let expected = sample_header();

crates/filemeta/src/metacache.rs

@@ -14,7 +14,7 @@
 
 use crate::{
     Error, FileInfo, FileInfoOpts, FileInfoVersions, FileMeta, FileMetaShallowVersion, Result, VersionType, get_file_info,
-    merge_file_meta_versions,
+    merge_file_meta_versions, merge_file_meta_versions_with_write_quorum,
 };
 use arc_swap::ArcSwapOption;
 use rmp::Marker;
@@ -334,7 +334,15 @@ impl MetaCacheEntries {
         &self.0
     }
 
-    pub fn resolve(&self, mut params: MetadataResolutionParams) -> Option<MetaCacheEntry> {
+    pub fn resolve(&self, params: MetadataResolutionParams) -> Option<MetaCacheEntry> {
+        self.resolve_inner(params, false)
+    }
+
+    pub fn resolve_with_write_quorum(&self, params: MetadataResolutionParams) -> Option<MetaCacheEntry> {
+        self.resolve_inner(params, true)
+    }
+
+    fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option<MetaCacheEntry> {
         if self.0.is_empty() {
             debug!(
                 bucket = %params.bucket,
@@ -412,7 +420,7 @@ impl MetaCacheEntries {
             return Some(selected);
         }
 
-        // If we would never be able to reach read quorum.
+        // If we would never be able to reach the required object quorum.
         if objs_valid < params.obj_quorum {
             debug!(
                 objs_valid,
@@ -423,13 +431,26 @@ impl MetaCacheEntries {
         }
 
         if objs_agree == objs_valid {
-            debug!(
-                selected = %selected.name,
-                objs_agree,
-                objs_valid,
-                "metacache resolve reused selected candidate because all valid object entries agreed"
-            );
-            return Some(selected);
+            let required_quorum = if enforce_write_quorum {
+                selected
+                    .cached
+                    .as_ref()
+                    .and_then(|cached| cached.versions.first())
+                    .map(|version| version.write_quorum(params.obj_quorum).max(params.obj_quorum))
+                    .unwrap_or(params.obj_quorum)
+            } else {
+                params.obj_quorum
+            };
+
+            if objs_agree >= required_quorum {
+                debug!(
+                    selected = %selected.name,
+                    objs_agree,
+                    objs_valid,
+                    "metacache resolve reused selected candidate because all valid object entries agreed"
+                );
+                return Some(selected);
+            }
         }
 
         let Some(cached) = selected.cached else {
@@ -437,7 +458,16 @@ impl MetaCacheEntries {
             return None;
         };
 
-        let versions = merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, &params.candidates);
+        let versions = if enforce_write_quorum {
+            merge_file_meta_versions_with_write_quorum(
+                params.obj_quorum,
+                params.strict,
+                params.requested_versions,
+                &params.candidates,
+            )
+        } else {
+            merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, &params.candidates)
+        };
         if versions.is_empty() {
             debug!(
                 selected = %selected.name,
@@ -1272,6 +1302,300 @@ mod tests {
         assert_ne!(extended_versions, cached.versions.len());
     }
 
+    fn metacache_entry_with_mod_time(mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry {
+        let mut metadata = HashMap::new();
+        metadata.insert("etag".to_string(), etag.to_string());
+
+        let mut meta = FileMeta::new();
+        meta.add_version(FileInfo {
+            volume: "bucket".to_string(),
+            name: "object".to_string(),
+            size: 1,
+            mod_time: Some(mod_time),
+            metadata,
+            ..Default::default()
+        })
+        .expect("test file metadata should accept object version");
+        let encoded = meta.marshal_msg().expect("test file metadata should marshal");
+
+        MetaCacheEntry {
+            name: "object".to_string(),
+            metadata: encoded,
+            cached: Some(meta),
+            reusable: false,
+        }
+    }
+
+    fn metacache_entry_with_erasure(
+        mod_time: OffsetDateTime,
+        etag: &str,
+        data_blocks: usize,
+        parity_blocks: usize,
+    ) -> MetaCacheEntry {
+        let mut metadata = HashMap::new();
+        metadata.insert("etag".to_string(), etag.to_string());
+
+        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
+        fi.volume = "bucket".to_string();
+        fi.name = "object".to_string();
+        fi.size = 1;
+        fi.mod_time = Some(mod_time);
+        fi.metadata = metadata;
+
+        let mut meta = FileMeta::new();
+        meta.add_version(fi).expect("test file metadata should accept object version");
+        let encoded = meta.marshal_msg().expect("test file metadata should marshal");
+
+        MetaCacheEntry {
+            name: "object".to_string(),
+            metadata: encoded,
+            cached: Some(meta),
+            reusable: false,
+        }
+    }
+
+    fn metacache_entry_with_erasure_versions(versions: &[(OffsetDateTime, &str, usize, usize)]) -> MetaCacheEntry {
+        let mut meta = FileMeta::new();
+        for (idx, (mod_time, etag, data_blocks, parity_blocks)) in versions.iter().enumerate() {
+            let mut metadata = HashMap::new();
+            metadata.insert("etag".to_string(), (*etag).to_string());
+
+            let mut fi = FileInfo::new("object", *data_blocks, *parity_blocks);
+            fi.volume = "bucket".to_string();
+            fi.name = "object".to_string();
+            let version_idx = u128::try_from(idx + 1).expect("test version index should fit u128");
+            fi.version_id = Some(Uuid::from_u128(version_idx));
+            fi.versioned = true;
+            fi.size = 1;
+            fi.mod_time = Some(*mod_time);
+            fi.metadata = metadata;
+
+            meta.add_version(fi).expect("test file metadata should accept object version");
+        }
+        let encoded = meta.marshal_msg().expect("test file metadata should marshal");
+
+        MetaCacheEntry {
+            name: "object".to_string(),
+            metadata: encoded,
+            cached: Some(meta),
+            reusable: false,
+        }
+    }
+
+    fn metacache_entry_without_header_ec(mut entry: MetaCacheEntry) -> MetaCacheEntry {
+        let mut cached = entry.cached.take().expect("test entry should have cached metadata");
+        for version in cached.versions.iter_mut() {
+            version.header.ec_m = 0;
+            version.header.ec_n = 0;
+        }
+        entry.metadata = cached.marshal_msg().expect("test file metadata should marshal");
+        entry.cached = Some(cached);
+        entry
+    }
+
+    fn metacache_dir_entry(name: &str) -> MetaCacheEntry {
+        MetaCacheEntry {
+            name: name.to_string(),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn resolve_rejects_partial_latest_and_returns_committed_previous_metadata() {
+        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let old_entry = metacache_entry_with_mod_time(old_mod_time, "old-etag");
+        let new_entry = metacache_entry_with_mod_time(new_mod_time, "new-etag");
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_entry.clone()),
+            Some(new_entry),
+            Some(old_entry.clone()),
+            Some(old_entry.clone()),
+            Some(old_entry.clone()),
+            Some(old_entry.clone()),
+            Some(old_entry),
+        ])
+        .resolve(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        })
+        .expect("previous committed metadata should still satisfy write quorum");
+
+        let info = resolved
+            .to_fileinfo("bucket")
+            .expect("resolved committed metadata should decode as file info");
+        assert_eq!(info.mod_time, Some(old_mod_time));
+        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
+    }
+
+    #[test]
+    fn resolve_rejects_low_parity_partial_latest_below_required_object_quorum() {
+        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 7, 1);
+        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1);
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry),
+            Some(old_entry.clone()),
+            Some(old_entry),
+        ])
+        .resolve_with_write_quorum(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        });
+
+        assert!(resolved.is_none());
+    }
+
+    #[test]
+    fn resolve_rejects_zero_parity_partial_latest_below_required_object_quorum() {
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 0);
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry),
+        ])
+        .resolve_with_write_quorum(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        });
+
+        assert!(resolved.is_none());
+    }
+
+    #[test]
+    fn resolve_accepts_low_parity_latest_at_required_object_quorum() {
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1);
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry.clone()),
+            Some(new_entry),
+        ])
+        .resolve_with_write_quorum(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        })
+        .expect("latest metadata should resolve after satisfying its own write quorum");
+
+        let info = resolved
+            .to_fileinfo("bucket")
+            .expect("resolved committed metadata should decode as file info");
+        assert_eq!(info.mod_time, Some(new_mod_time));
+        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("new-etag"));
+    }
+
+    #[test]
+    fn resolve_skips_low_parity_partial_latest_and_returns_committed_previous_version() {
+        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4);
+        let new_and_old_entry =
+            metacache_entry_with_erasure_versions(&[(old_mod_time, "old-etag", 4, 4), (new_mod_time, "new-etag", 7, 1)]);
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry),
+            Some(old_entry.clone()),
+            Some(old_entry),
+        ])
+        .resolve_with_write_quorum(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        })
+        .expect("previous committed metadata should resolve after rejecting partial latest");
+
+        let info = resolved
+            .to_fileinfo("bucket")
+            .expect("resolved committed metadata should decode as file info");
+        assert_eq!(info.mod_time, Some(old_mod_time));
+        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
+    }
+
+    #[test]
+    fn resolve_skips_legacy_header_low_parity_partial_latest_using_payload_quorum() {
+        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
+        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4);
+        let new_and_old_entry = metacache_entry_without_header_ec(metacache_entry_with_erasure_versions(&[
+            (old_mod_time, "old-etag", 4, 4),
+            (new_mod_time, "new-etag", 7, 1),
+        ]));
+
+        let resolved = MetaCacheEntries(vec![
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry.clone()),
+            Some(new_and_old_entry),
+            Some(old_entry.clone()),
+            Some(old_entry),
+        ])
+        .resolve_with_write_quorum(MetadataResolutionParams {
+            obj_quorum: 5,
+            requested_versions: 1,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        })
+        .expect("previous committed metadata should resolve after rejecting legacy-header partial latest");
+
+        let info = resolved
+            .to_fileinfo("bucket")
+            .expect("resolved committed metadata should decode as file info");
+        assert_eq!(info.mod_time, Some(old_mod_time));
+        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
+    }
+
+    #[test]
+    fn resolve_rejects_partial_directory_below_dir_quorum() {
+        let partial_dir = metacache_dir_entry("prefix/");
+
+        let resolved = MetaCacheEntries(vec![Some(partial_dir.clone()), Some(partial_dir)]).resolve(MetadataResolutionParams {
+            dir_quorum: 5,
+            obj_quorum: 5,
+            bucket: "bucket".to_string(),
+            strict: true,
+            ..Default::default()
+        });
+
+        assert!(resolved.is_none());
+    }
+
     fn build_hashmap_cache(update_size: usize) -> Arc<Cache<HashMap<usize, usize>>> {
         let generation = Arc::new(AtomicUsize::new(0));
         Arc::new(Cache::new(