fix: support legacy bucket metadata decoding (#2227)

crates/ecstore/src/bucket/metadata.rs

@@ -33,7 +33,7 @@ use serde::Serializer;
 use std::collections::HashMap;
 use std::io::{Read, Write};
 use std::sync::Arc;
-use time::OffsetDateTime;
+use time::{Date, OffsetDateTime, PrimitiveDateTime, Time as CivilTime, UtcOffset};
 use tracing::error;
 
 fn read_msgp_str<R: Read>(rd: &mut R) -> Result<String> {
@@ -43,19 +43,177 @@ fn read_msgp_str<R: Read>(rd: &mut R) -> Result<String> {
     Ok(String::from_utf8(buf)?)
 }
 
+fn read_msgp_bool<R: Read>(rd: &mut R) -> Result<bool> {
+    let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    match marker {
+        rmp::Marker::True => Ok(true),
+        rmp::Marker::False => Ok(false),
+        rmp::Marker::FixPos(v) => Ok(v != 0),
+        rmp::Marker::U8 => Ok(read_u8(rd)? != 0),
+        rmp::Marker::U16 => Ok(read_u16_raw(rd)? != 0),
+        rmp::Marker::U32 => Ok(read_u32_raw(rd)? != 0),
+        rmp::Marker::U64 => Ok(read_u64_raw(rd)? != 0),
+        rmp::Marker::I8 => Ok(read_i8_raw(rd)? != 0),
+        rmp::Marker::I16 => Ok(read_i16_raw(rd)? != 0),
+        rmp::Marker::I32 => Ok(read_i32_raw(rd)? != 0),
+        rmp::Marker::I64 => Ok(read_i64_raw(rd)? != 0),
+        rmp::Marker::FixNeg(v) => Ok(v != 0),
+        _ => Err(Error::other(format!("expected bool or int-like bool, got marker: {marker:?}"))),
+    }
+}
+
 fn read_msgp_time_value<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
     let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?;
     match marker {
         rmp::Marker::Null => Ok(OffsetDateTime::UNIX_EPOCH),
         rmp::Marker::Ext8 => read_msgp_ext8_time(rd),
+        rmp::Marker::FixArray(len) => read_msgp_legacy_compact_time(rd, u32::from(len)),
+        rmp::Marker::Array16 => {
+            let len = read_u16_raw(rd)?;
+            read_msgp_legacy_compact_time(rd, u32::from(len))
+        }
+        rmp::Marker::Array32 => {
+            let len = read_u32_raw(rd)?;
+            read_msgp_legacy_compact_time(rd, len)
+        }
+        rmp::Marker::Bin8 => {
+            let len = usize::from(read_u8(rd)?);
+            read_msgp_time_value_from_embedded_bin(rd, len)
+        }
+        rmp::Marker::Bin16 => {
+            let len = usize::from(read_u16_raw(rd)?);
+            read_msgp_time_value_from_embedded_bin(rd, len)
+        }
+        rmp::Marker::Bin32 => {
+            let len = read_u32_raw(rd)? as usize;
+            read_msgp_time_value_from_embedded_bin(rd, len)
+        }
         _ => Err(Error::other(format!("expected time ext or nil, got marker: {marker:?}"))),
     }
 }
 
-fn read_msgp_bin<R: Read>(rd: &mut R) -> Result<Vec<u8>> {
-    let len = rmp::decode::read_bin_len(rd)? as usize;
+fn read_u8<R: Read>(rd: &mut R) -> Result<u8> {
+    let mut buf = [0u8; 1];
+    rd.read_exact(&mut buf)?;
+    Ok(buf[0])
+}
+
+fn read_u16_raw<R: Read>(rd: &mut R) -> Result<u16> {
+    let mut buf = [0u8; 2];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_u16(&buf))
+}
+
+fn read_u32_raw<R: Read>(rd: &mut R) -> Result<u32> {
+    let mut buf = [0u8; 4];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_u32(&buf))
+}
+
+fn read_u64_raw<R: Read>(rd: &mut R) -> Result<u64> {
+    let mut buf = [0u8; 8];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_u64(&buf))
+}
+
+fn read_i8_raw<R: Read>(rd: &mut R) -> Result<i8> {
+    Ok(read_u8(rd)? as i8)
+}
+
+fn read_i16_raw<R: Read>(rd: &mut R) -> Result<i16> {
+    let mut buf = [0u8; 2];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_i16(&buf))
+}
+
+fn read_i32_raw<R: Read>(rd: &mut R) -> Result<i32> {
+    let mut buf = [0u8; 4];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_i32(&buf))
+}
+
+fn read_i64_raw<R: Read>(rd: &mut R) -> Result<i64> {
+    let mut buf = [0u8; 8];
+    rd.read_exact(&mut buf)?;
+    Ok(BigEndian::read_i64(&buf))
+}
+
+fn read_msgp_time_value_from_embedded_bin<R: Read>(rd: &mut R, len: usize) -> Result<OffsetDateTime> {
     let mut buf = vec![0u8; len];
     rd.read_exact(&mut buf)?;
+    let mut cur = std::io::Cursor::new(buf);
+    read_msgp_time_value(&mut cur)
+}
+
+fn read_msgp_legacy_compact_time<R: Read>(rd: &mut R, len: u32) -> Result<OffsetDateTime> {
+    if len != 9 {
+        return Err(Error::other(format!("invalid legacy compact time len: {len}")));
+    }
+
+    let year: i32 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let ordinal: u16 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let hour: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let minute: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let second: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let nanosecond: u32 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let offset_hour: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let offset_minute: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    let offset_second: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+
+    let date =
+        Date::from_ordinal_date(year, ordinal).map_err(|e| Error::other(format!("invalid legacy compact time date: {e}")))?;
+    let time = CivilTime::from_hms_nano(hour, minute, second, nanosecond)
+        .map_err(|e| Error::other(format!("invalid legacy compact time time: {e}")))?;
+    let offset = UtcOffset::from_hms(offset_hour, offset_minute, offset_second)
+        .map_err(|e| Error::other(format!("invalid legacy compact time offset: {e}")))?;
+
+    Ok(PrimitiveDateTime::new(date, time)
+        .assume_offset(offset)
+        .to_offset(UtcOffset::UTC))
+}
+
+fn read_msgp_bin<R: Read>(rd: &mut R) -> Result<Vec<u8>> {
+    let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+    match marker {
+        rmp::Marker::Null => Ok(Vec::new()),
+        rmp::Marker::Bin8 => {
+            let len = usize::from(read_u8(rd)?);
+            read_exact_bytes(rd, len)
+        }
+        rmp::Marker::Bin16 => {
+            let len = usize::from(read_u16_raw(rd)?);
+            read_exact_bytes(rd, len)
+        }
+        rmp::Marker::Bin32 => {
+            let len = read_u32_raw(rd)? as usize;
+            read_exact_bytes(rd, len)
+        }
+        rmp::Marker::FixArray(len) => read_msgp_legacy_byte_array(rd, u32::from(len)),
+        rmp::Marker::Array16 => {
+            let len = read_u16_raw(rd)?;
+            read_msgp_legacy_byte_array(rd, u32::from(len))
+        }
+        rmp::Marker::Array32 => {
+            let len = read_u32_raw(rd)?;
+            read_msgp_legacy_byte_array(rd, len)
+        }
+        _ => Err(Error::other(format!("expected bin or byte array, got marker: {marker:?}"))),
+    }
+}
+
+fn read_exact_bytes<R: Read>(rd: &mut R, len: usize) -> Result<Vec<u8>> {
+    let mut buf = vec![0u8; len];
+    rd.read_exact(&mut buf)?;
+    Ok(buf)
+}
+
+fn read_msgp_legacy_byte_array<R: Read>(rd: &mut R, len: u32) -> Result<Vec<u8>> {
+    let mut buf = Vec::with_capacity(len as usize);
+    for _ in 0..len {
+        let value: i64 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?;
+        let byte = u8::try_from(value).map_err(|_| Error::other(format!("byte value out of range: {value}")))?;
+        buf.push(byte);
+    }
     Ok(buf)
 }
 
@@ -227,18 +385,20 @@ impl BucketMetadata {
             match key.as_str() {
                 "Name" => self.name = read_msgp_str(rd)?,
                 "Created" => self.created = read_msgp_time_value(rd)?,
-                "LockEnabled" => self.lock_enabled = rmp::decode::read_bool(rd)?,
-                "PolicyConfigJSON" => self.policy_config_json = read_msgp_bin(rd)?,
-                "NotificationConfigXML" => self.notification_config_xml = read_msgp_bin(rd)?,
-                "LifecycleConfigXML" => self.lifecycle_config_xml = read_msgp_bin(rd)?,
-                "ObjectLockConfigXML" => self.object_lock_config_xml = read_msgp_bin(rd)?,
-                "VersioningConfigXML" => self.versioning_config_xml = read_msgp_bin(rd)?,
-                "EncryptionConfigXML" => self.encryption_config_xml = read_msgp_bin(rd)?,
-                "TaggingConfigXML" => self.tagging_config_xml = read_msgp_bin(rd)?,
-                "QuotaConfigJSON" => self.quota_config_json = read_msgp_bin(rd)?,
-                "ReplicationConfigXML" => self.replication_config_xml = read_msgp_bin(rd)?,
-                "BucketTargetsConfigJSON" => self.bucket_targets_config_json = read_msgp_bin(rd)?,
-                "BucketTargetsConfigMetaJSON" => self.bucket_targets_config_meta_json = read_msgp_bin(rd)?,
+                "LockEnabled" => self.lock_enabled = read_msgp_bool(rd)?,
+                "PolicyConfigJSON" | "PolicyConfigJson" => self.policy_config_json = read_msgp_bin(rd)?,
+                "NotificationConfigXML" | "NotificationConfigXml" => self.notification_config_xml = read_msgp_bin(rd)?,
+                "LifecycleConfigXML" | "LifecycleConfigXml" => self.lifecycle_config_xml = read_msgp_bin(rd)?,
+                "ObjectLockConfigXML" | "ObjectLockConfigXml" => self.object_lock_config_xml = read_msgp_bin(rd)?,
+                "VersioningConfigXML" | "VersioningConfigXml" => self.versioning_config_xml = read_msgp_bin(rd)?,
+                "EncryptionConfigXML" | "EncryptionConfigXml" => self.encryption_config_xml = read_msgp_bin(rd)?,
+                "TaggingConfigXML" | "TaggingConfigXml" => self.tagging_config_xml = read_msgp_bin(rd)?,
+                "QuotaConfigJSON" | "QuotaConfigJson" => self.quota_config_json = read_msgp_bin(rd)?,
+                "ReplicationConfigXML" | "ReplicationConfigXml" => self.replication_config_xml = read_msgp_bin(rd)?,
+                "BucketTargetsConfigJSON" | "BucketTargetsConfigJson" => self.bucket_targets_config_json = read_msgp_bin(rd)?,
+                "BucketTargetsConfigMetaJSON" | "BucketTargetsConfigMetaJson" => {
+                    self.bucket_targets_config_meta_json = read_msgp_bin(rd)?
+                }
                 "PolicyConfigUpdatedAt" => self.policy_config_updated_at = read_msgp_time_value(rd)?,
                 "ObjectLockConfigUpdatedAt" => self.object_lock_config_updated_at = read_msgp_time_value(rd)?,
                 "EncryptionConfigUpdatedAt" => self.encryption_config_updated_at = read_msgp_time_value(rd)?,
@@ -250,9 +410,11 @@ impl BucketMetadata {
                 "NotificationConfigUpdatedAt" => self.notification_config_updated_at = read_msgp_time_value(rd)?,
                 "BucketTargetsConfigUpdatedAt" => self.bucket_targets_config_updated_at = read_msgp_time_value(rd)?,
                 "BucketTargetsConfigMetaUpdatedAt" => self.bucket_targets_config_meta_updated_at = read_msgp_time_value(rd)?,
-                "CorsConfigXML" => self.cors_config_xml = read_msgp_bin(rd)?,
-                "PublicAccessBlockConfigXML" => self.public_access_block_config_xml = read_msgp_bin(rd)?,
-                "BucketAclConfigJSON" => self.bucket_acl_config_json = read_msgp_bin(rd)?,
+                "CorsConfigXML" | "CorsConfigXml" => self.cors_config_xml = read_msgp_bin(rd)?,
+                "PublicAccessBlockConfigXML" | "PublicAccessBlockConfigXml" => {
+                    self.public_access_block_config_xml = read_msgp_bin(rd)?
+                }
+                "BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?,
                 "CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?,
                 "PublicAccessBlockConfigUpdatedAt" => self.public_access_block_config_updated_at = read_msgp_time_value(rd)?,
                 "BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?,

crates/ecstore/src/bucket/metadata_test.rs

@@ -98,6 +98,143 @@ async fn unmarshal_test_bucket_metadata() {
     assert!(bm.bucket_acl_config_json.is_empty());
 }
 
+#[test]
+fn unmarshal_legacy_compact_time_bucket_metadata() {
+    use faster_hex::hex_decode;
+
+    let legacy_hex = concat!(
+        "83",
+        "a44e616d65",
+        "a474657374",
+        "a743726561746564",
+        "99cd07e9cd01100c1021ce2026b1fa000000",
+        "ab4c6f636b456e61626c6564",
+        "c2"
+    );
+
+    let mut bytes = vec![0u8; legacy_hex.len() / 2];
+    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");
+
+    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy compact time should decode");
+
+    assert_eq!(bm.name, "test");
+    assert_eq!(bm.created.unix_timestamp(), 1759148193);
+    assert_eq!(bm.created.nanosecond(), 539406842);
+    assert!(!bm.lock_enabled);
+}
+
+#[test]
+fn unmarshal_bin_wrapped_ext_time_bucket_metadata() {
+    use faster_hex::hex_decode;
+
+    let wrapped_hex = concat!(
+        "83",
+        "a44e616d65",
+        "a464616461",
+        "a743726561746564",
+        "c40fc70c05fffffff1886e090000000000",
+        "ab4c6f636b456e61626c6564",
+        "c2"
+    );
+
+    let mut bytes = vec![0u8; wrapped_hex.len() / 2];
+    hex_decode(wrapped_hex.as_bytes(), &mut bytes).expect("valid hex");
+
+    let bm = BucketMetadata::unmarshal(&bytes).expect("bin-wrapped ext time should decode");
+
+    assert_eq!(bm.name, "dada");
+    assert_eq!(bm.created.unix_timestamp(), -62135596800);
+    assert!(!bm.lock_enabled);
+}
+
+#[test]
+fn unmarshal_legacy_rmp_serde_field_aliases_and_byte_arrays() {
+    use faster_hex::hex_decode;
+
+    let legacy_hex = concat!(
+        "85",
+        "a44e616d65",
+        "a474657374",
+        "a743726561746564",
+        "99cd07e9cd01100c1021ce2026b1fa000000",
+        "ab4c6f636b456e61626c6564",
+        "c2",
+        "b0506f6c696379436f6e6669674a736f6e",
+        "93010203",
+        "bb4275636b657454617267657473436f6e6669674d6574614a736f6e",
+        "920405"
+    );
+
+    let mut bytes = vec![0u8; legacy_hex.len() / 2];
+    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");
+
+    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy field aliases and byte arrays should decode");
+
+    assert_eq!(bm.name, "test");
+    assert_eq!(bm.created.unix_timestamp(), 1759148193);
+    assert_eq!(bm.created.nanosecond(), 539406842);
+    assert_eq!(bm.policy_config_json, vec![1, 2, 3]);
+    assert_eq!(bm.bucket_targets_config_meta_json, vec![4, 5]);
+    assert!(!bm.lock_enabled);
+}
+
+#[test]
+fn unmarshal_legacy_bin16_and_array16_bucket_metadata() {
+    let policy = vec![b'x'; 257];
+    let targets_meta: Vec<u8> = (0u8..=16).collect();
+
+    let mut bytes = Vec::new();
+    rmp::encode::write_map_len(&mut bytes, 5).unwrap();
+    rmp::encode::write_str(&mut bytes, "Name").unwrap();
+    rmp::encode::write_str(&mut bytes, "test-bucket").unwrap();
+    rmp::encode::write_str(&mut bytes, "Created").unwrap();
+    bytes.extend_from_slice(&[
+        0xc7, 0x0c, 0x05, 0x00, 0x00, 0x00, 0x00, 0x65, 0x92, 0x00, 0x80, 0x00, 0x00, 0x00, 0x00,
+    ]);
+    rmp::encode::write_str(&mut bytes, "LockEnabled").unwrap();
+    bytes.push(0x01);
+    rmp::encode::write_str(&mut bytes, "PolicyConfigJson").unwrap();
+    rmp::encode::write_bin(&mut bytes, &policy).unwrap();
+    rmp::encode::write_str(&mut bytes, "BucketTargetsConfigMetaJson").unwrap();
+    rmp::encode::write_array_len(&mut bytes, targets_meta.len() as u32).unwrap();
+    for byte in &targets_meta {
+        rmp::encode::write_uint(&mut bytes, u64::from(*byte)).unwrap();
+    }
+
+    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy bin16 and array16 should decode");
+
+    assert_eq!(bm.name, "test-bucket");
+    assert_eq!(bm.created.unix_timestamp(), 1704067200);
+    assert!(bm.lock_enabled);
+    assert_eq!(bm.policy_config_json, policy);
+    assert_eq!(bm.bucket_targets_config_meta_json, targets_meta);
+}
+
+#[test]
+fn unmarshal_legacy_numeric_bool_bucket_metadata() {
+    use faster_hex::hex_decode;
+
+    let legacy_hex = concat!(
+        "83",
+        "a44e616d65",
+        "ab746573742d6275636b6574",
+        "a743726561746564",
+        "c70c05000000006592008000000000",
+        "ab4c6f636b456e61626c6564",
+        "01"
+    );
+
+    let mut bytes = vec![0u8; legacy_hex.len() / 2];
+    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");
+
+    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy numeric bool should decode");
+
+    assert_eq!(bm.name, "test-bucket");
+    assert_eq!(bm.created.unix_timestamp(), 1704067200);
+    assert_eq!(bm.created.nanosecond(), 0);
+    assert!(bm.lock_enabled);
+}
+
 #[tokio::test]
 async fn marshal_msg_complete_example() {
     // Create a complete BucketMetadata with various configurations

crates/filemeta/src/filemeta.rs

@@ -74,6 +74,14 @@ pub(crate) fn data_key_for_version(version_id: Option<Uuid>) -> String {
     }
 }
 
+fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
+    if version_id.is_none() || version_id == Some(Uuid::nil()) {
+        Some(Uuid::nil().to_string())
+    } else {
+        None
+    }
+}
+
 pub const TRANSITION_COMPLETE: &str = "complete";
 pub const TRANSITION_PENDING: &str = "pending";
 
@@ -164,6 +172,21 @@ impl FileMeta {
         });
     }
 
+    fn find_inline_data_for_version(&self, version_id: Option<Uuid>) -> Result<Option<Vec<u8>>> {
+        let key = data_key_for_version(version_id);
+        if let Some(data) = self.data.find(key.as_str())? {
+            return Ok(Some(data));
+        }
+
+        if let Some(legacy_key) = legacy_data_key_for_version(version_id)
+            && legacy_key != key
+        {
+            return self.data.find(legacy_key.as_str());
+        }
+
+        Ok(None)
+    }
+
     // Find version
     pub fn find_version(&self, vid: Option<Uuid>) -> Result<(usize, FileMetaVersion)> {
         let vid = vid.unwrap_or_default();
@@ -709,10 +732,7 @@ impl FileMeta {
             }
 
             if read_data && fi.inline_data() {
-                fi.data = self
-                    .data
-                    .find(data_key_for_version(fi.version_id).as_str())?
-                    .map(bytes::Bytes::from);
+                fi.data = self.find_inline_data_for_version(fi.version_id)?.map(bytes::Bytes::from);
             }
 
             found_fi = Some(fi);
@@ -1071,6 +1091,28 @@ mod test {
         assert!(!inline_data.is_empty(), "Inline data should not be empty");
     }
 
+    #[test]
+    fn test_into_fileinfo_reads_legacy_nil_uuid_inline_key() {
+        let mut fm = FileMeta::new();
+        let mut fi = FileInfo::new("test", 2, 1);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.data = Some(Bytes::from_static(b"legacy-inline"));
+        fi.set_inline_data();
+        fm.add_version(fi).unwrap();
+
+        let current_key = data_key_for_version(Some(Uuid::nil()));
+        let legacy_key = legacy_data_key_for_version(Some(Uuid::nil())).unwrap();
+        let payload = fm.data.find(current_key.as_str()).unwrap().unwrap();
+
+        fm.data.remove_key(current_key.as_str()).unwrap();
+        fm.data.replace(legacy_key.as_str(), payload.clone()).unwrap();
+
+        let restored = fm.into_fileinfo("bucket", "test", "", true, false, true).unwrap();
+
+        assert!(restored.inline_data());
+        assert_eq!(restored.data, Some(Bytes::from(payload)));
+    }
+
     #[test]
     fn test_error_handling_and_recovery() {
         // Test error handling and recovery

crates/filemeta/src/filemeta/inline_data.rs

@@ -43,11 +43,7 @@ impl FileMeta {
         let vid = version_id.unwrap_or_default();
 
         if self.data.entries().unwrap_or_default() > 0
-            && self
-                .data
-                .find(super::data_key_for_version(version_id).as_str())
-                .unwrap_or_default()
-                .is_some()
+            && self.find_inline_data_for_version(version_id).unwrap_or_default().is_some()
         {
             return 0;
         }