rustfs_filemeta/
fileinfo.rs

1// Copyright 2024 RustFS Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use crate::metadata_keys;
16use crate::{Error, ReplicationState, ReplicationStatusType, Result, TRANSITION_COMPLETE, VersionPurgeStatusType};
17use bytes::Bytes;
18use rmp_serde::Serializer;
19use rustfs_utils::HashAlgorithm;
20use rustfs_utils::http::{
21    AMZ_OBJECT_TAGGING, SUFFIX_COMPRESSION, SUFFIX_DATA_MOVED, SUFFIX_DATA_MOVED_TAGS, SUFFIX_FREE_VERSION, SUFFIX_HEALING,
22    SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID,
23    contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key,
24    starts_with_ignore_ascii_case,
25};
26use serde::de::{self, MapAccess, SeqAccess, Visitor, value::MapAccessDeserializer};
27use serde::ser::SerializeMap;
28use serde::{Deserialize, Serialize};
29use std::collections::HashMap;
30use time::{OffsetDateTime, PrimitiveDateTime, format_description::well_known::Rfc3339};
31use time::{format_description::FormatItem, macros::format_description};
32use uuid::Uuid;
33
34pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
35pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M
36
37pub(crate) const MAX_ERASURE_SHARDS: usize = 16;
38const MAX_FILEINFO_PARTS: usize = 10_000;
39const MAX_FILEINFO_CHECKSUMS: usize = 10_000;
40const FILEINFO_PART_BITMAP_WORD_BITS: usize = std::mem::size_of::<u64>() * 8;
41const FILEINFO_PART_BITMAP_WORDS: usize = MAX_FILEINFO_PARTS.div_ceil(FILEINFO_PART_BITMAP_WORD_BITS);
42
43// Additional constants from Go version
44// Intentionally duplicated (S3 wire literal): rustfs-replication and
45// rustfs-object-data-cache carry their own independent "null" constants so
46// they stay free of a rustfs-filemeta dependency. Keep all three in sync.
47pub const NULL_VERSION_ID: &str = "null";
48// pub const RUSTFS_ERASURE_UPGRADED: &str = "x-rustfs-internal-erasure-upgraded";
49
50pub const TIER_FV_ID: &str = "tier-free-versionID";
51pub const TIER_FV_MARKER: &str = "tier-free-marker";
52pub const TIER_SKIP_FV_ID: &str = "tier-skip-fvid";
53
54const ERR_RESTORE_HDR_MALFORMED: &str = "x-amz-restore header malformed";
55
56const RFC1123: &[FormatItem<'_>] =
57    format_description!("[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT");
58
59#[derive(Deserialize, Debug, PartialEq, Clone, Default)]
60pub struct ObjectPartInfo {
61    pub etag: String,
62    pub number: usize,
63    pub size: usize,
64    pub actual_size: i64, // Original data size
65    pub mod_time: Option<OffsetDateTime>,
66    // Index holds the index of the part in the erasure coding
67    pub index: Option<Bytes>,
68    // Checksums holds checksums of the part
69    pub checksums: Option<HashMap<String, String>>,
70    pub error: Option<String>,
71    #[serde(default)]
72    pub integrity: Option<crate::shard_integrity::PartIntegrity>,
73}
74
75impl Serialize for ObjectPartInfo {
76    fn serialize<S: serde::Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
77        // Named fields let release readers ignore optional extensions. Appending
78        // a ninth element to the old positional array makes those readers fail.
79        let mut map = serializer.serialize_map(Some(8 + usize::from(self.integrity.is_some())))?;
80        map.serialize_entry("etag", &self.etag)?;
81        map.serialize_entry("number", &self.number)?;
82        map.serialize_entry("size", &self.size)?;
83        map.serialize_entry("actual_size", &self.actual_size)?;
84        map.serialize_entry("mod_time", &self.mod_time)?;
85        map.serialize_entry("index", &self.index)?;
86        map.serialize_entry("checksums", &self.checksums)?;
87        map.serialize_entry("error", &self.error)?;
88        if let Some(integrity) = &self.integrity {
89            map.serialize_entry("integrity", integrity)?;
90        }
91        map.end()
92    }
93}
94
95impl ObjectPartInfo {
96    pub fn marshal_msg(&self) -> Result<Vec<u8>> {
97        let mut buf = Vec::new();
98        self.serialize(&mut Serializer::new(&mut buf))?;
99        Ok(buf)
100    }
101
102    pub fn unmarshal(buf: &[u8]) -> Result<Self> {
103        let t: ObjectPartInfo = rmp_serde::from_slice(buf)?;
104        Ok(t)
105    }
106}
107
108#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
109// ChecksumInfo - carries checksums of individual scattered parts per disk.
110pub struct ChecksumInfo {
111    pub part_number: usize,
112    pub algorithm: HashAlgorithm,
113    pub hash: Bytes,
114}
115
116#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Default, Clone)]
117pub enum ErasureAlgo {
118    #[default]
119    Invalid = 0,
120    ReedSolomon = 1,
121}
122
123impl ErasureAlgo {
124    pub fn valid(&self) -> bool {
125        *self > ErasureAlgo::Invalid
126    }
127    pub fn to_u8(&self) -> u8 {
128        match self {
129            ErasureAlgo::Invalid => 0,
130            ErasureAlgo::ReedSolomon => 1,
131        }
132    }
133
134    pub fn from_u8(u: u8) -> Self {
135        match u {
136            1 => ErasureAlgo::ReedSolomon,
137            _ => ErasureAlgo::Invalid,
138        }
139    }
140}
141
142impl std::fmt::Display for ErasureAlgo {
143    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
144        match self {
145            ErasureAlgo::Invalid => write!(f, "Invalid"),
146            ErasureAlgo::ReedSolomon => write!(f, "{ERASURE_ALGORITHM}"),
147        }
148    }
149}
150
151#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
152// ErasureInfo holds erasure coding and bitrot related information.
153pub struct ErasureInfo {
154    // Algorithm is the String representation of erasure-coding-algorithm
155    pub algorithm: String,
156    // DataBlocks is the number of data blocks for erasure-coding
157    pub data_blocks: usize,
158    // ParityBlocks is the number of parity blocks for erasure-coding
159    pub parity_blocks: usize,
160    // BlockSize is the size of one erasure-coded block
161    pub block_size: usize,
162    // Index is the index of the current disk
163    pub index: usize,
164    // Distribution is the distribution of the data and parity blocks
165    pub distribution: Vec<usize>,
166    // Checksums holds all bitrot checksums of all erasure encoded blocks
167    pub checksums: Vec<ChecksumInfo>,
168}
169
170pub fn calc_shard_size(block_size: usize, data_shards: usize) -> usize {
171    (block_size.div_ceil(data_shards) + 1) & !1
172}
173
174fn checked_calc_shard_size(block_size: usize, data_shards: usize) -> Option<usize> {
175    if data_shards == 0 {
176        return None;
177    }
178
179    block_size.div_ceil(data_shards).checked_add(1).map(|size| size & !1)
180}
181
182impl ErasureInfo {
183    pub fn get_checksum_info(&self, part_number: usize) -> ChecksumInfo {
184        for sum in &self.checksums {
185            if sum.part_number == part_number {
186                return sum.clone();
187            }
188        }
189
190        ChecksumInfo {
191            algorithm: HashAlgorithm::HighwayHash256S,
192            ..Default::default()
193        }
194    }
195
196    /// Calculate the size of each shard.
197    pub fn shard_size(&self) -> usize {
198        calc_shard_size(self.block_size, self.data_blocks)
199    }
200    /// Calculate the total erasure file size for a given original size.
201    // Returns the final erasure size from the original size
202    pub fn shard_file_size(&self, total_length: i64) -> i64 {
203        if total_length == 0 {
204            return 0;
205        }
206
207        if total_length < 0 {
208            return total_length;
209        }
210
211        let total_length = total_length as usize;
212
213        let num_shards = total_length / self.block_size;
214        let last_block_size = total_length % self.block_size;
215        let last_shard_size = calc_shard_size(last_block_size, self.data_blocks);
216        (num_shards * self.shard_size() + last_shard_size) as i64
217    }
218
219    /// Check if this ErasureInfo equals another ErasureInfo
220    pub fn equals(&self, other: &ErasureInfo) -> bool {
221        if self.algorithm != other.algorithm {
222            return false;
223        }
224
225        if self.data_blocks != other.data_blocks {
226            return false;
227        }
228
229        if self.parity_blocks != other.parity_blocks {
230            return false;
231        }
232
233        if self.block_size != other.block_size {
234            return false;
235        }
236
237        if self.distribution.len() != other.distribution.len() {
238            return false;
239        }
240        for (i, v) in self.distribution.iter().enumerate() {
241            if v != &other.distribution[i] {
242                return false;
243            }
244        }
245        true
246    }
247}
248
249// #[derive(Debug, Clone)]
250#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone, Copy, Default)]
251#[serde(rename_all = "kebab-case")]
252pub enum TransitionVersionState {
253    #[default]
254    Unknown,
255    KnownDisabled,
256    SuspendedNull,
257    Exact,
258}
259
260impl TransitionVersionState {
261    pub const fn as_str(self) -> &'static str {
262        match self {
263            Self::Unknown => "unknown",
264            Self::KnownDisabled => "known-disabled",
265            Self::SuspendedNull => "suspended-null",
266            Self::Exact => "exact",
267        }
268    }
269}
270
271#[derive(PartialEq, Clone, Default)]
272pub struct FileInfo {
273    pub volume: String,
274    pub name: String,
275    pub version_id: Option<Uuid>,
276    pub is_latest: bool,
277    pub deleted: bool,
278    pub transition_status: String,
279    pub transitioned_objname: String,
280    pub transition_tier: String,
281    pub transition_version_id: Option<Uuid>,
282    pub transition_version: Option<String>,
283    pub transition_version_state: TransitionVersionState,
284    pub expire_restored: bool,
285    pub data_dir: Option<Uuid>,
286    pub mod_time: Option<OffsetDateTime>,
287    pub size: i64,
288    // File mode bits
289    pub mode: Option<u32>,
290    // WrittenByVersion is the unix time stamp of the version that created this version of the object
291    pub written_by_version: Option<u64>,
292    pub metadata: HashMap<String, String>,
293    pub parts: Vec<ObjectPartInfo>,
294    pub erasure: ErasureInfo,
295    // MarkDeleted marks this version as deleted
296    pub mark_deleted: bool,
297    // ReplicationState - Internal replication state to be passed back in ObjectInfo
298    pub replication_state_internal: Option<ReplicationState>,
299    pub data: Option<Bytes>,
300    pub num_versions: usize,
301    pub successor_mod_time: Option<OffsetDateTime>,
302    pub fresh: bool,
303    pub idx: usize,
304    // Combined checksum when object was uploaded
305    pub checksum: Option<Bytes>,
306    pub versioned: bool,
307    /// True when version meta was parsed via rmp_serde fallback (legacy format).
308    pub uses_legacy_checksum: bool,
309    /// Transient PUT intent for the cleanup owner of a replaced tiered null version.
310    /// Named-map RPC readers may ignore this field; xl.meta never persists it.
311    pub overwrite_tier_free_version_id: Option<Uuid>,
312}
313
314/// Metadata keys whose values carry sealed encryption material (KEK-wrapped DEK,
315/// IV) under either the `x-rustfs-internal-` or `x-minio-internal-` prefix.
316/// Values of these keys must never reach logs at any level.
317fn is_sensitive_metadata_key(key: &str) -> bool {
318    // `is_encryption_metadata_key` covers the x-minio-internal- SSE prefix but not
319    // its reserved x-rustfs-internal- twin, which has no writer today but must
320    // stay redacted in case one appears.
321    is_encryption_metadata_key(key)
322        || starts_with_ignore_ascii_case(key, rustfs_utils::http::RUSTFS_INTERNAL_ENCRYPTION_PREFIX)
323        || rustfs_utils::http::REPLICATION_SSE_TRANSPORT_PREFIXES
324            .iter()
325            .any(|prefix| starts_with_ignore_ascii_case(key, prefix))
326}
327
328struct RedactedMetadata<'a>(&'a HashMap<String, String>);
329
330impl std::fmt::Debug for RedactedMetadata<'_> {
331    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
332        let mut map = f.debug_map();
333        for (key, value) in self.0 {
334            if is_sensitive_metadata_key(key) {
335                map.entry(key, &format_args!("<redacted {} bytes>", value.len()));
336            } else {
337                map.entry(key, value);
338            }
339        }
340        map.finish()
341    }
342}
343
344struct ElidedBytes<'a>(&'a Option<Bytes>);
345
346impl std::fmt::Debug for ElidedBytes<'_> {
347    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
348        match self.0 {
349            Some(bytes) => write!(f, "Some(<{} bytes elided>)", bytes.len()),
350            None => f.write_str("None"),
351        }
352    }
353}
354
355// Manual Debug: `data` holds full inline object bytes (plaintext user content for
356// non-SSE objects) and `metadata` holds sealed key material — both must stay out
357// of Debug output so whole-struct log dumps cannot leak them. `checksum` is elided
358// too: for non-SSE objects it fingerprints plaintext content, and its raw bytes
359// carry no diagnostic value. The exhaustive destructuring (no `..`) forces every
360// future field through an explicit show/redact decision here.
361impl std::fmt::Debug for FileInfo {
362    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
363        let Self {
364            volume,
365            name,
366            version_id,
367            is_latest,
368            deleted,
369            transition_status,
370            transitioned_objname,
371            transition_tier,
372            transition_version_id,
373            transition_version,
374            transition_version_state,
375            expire_restored,
376            data_dir,
377            mod_time,
378            size,
379            mode,
380            written_by_version,
381            metadata,
382            parts,
383            erasure,
384            mark_deleted,
385            replication_state_internal,
386            data,
387            num_versions,
388            successor_mod_time,
389            fresh,
390            idx,
391            checksum,
392            versioned,
393            uses_legacy_checksum,
394            overwrite_tier_free_version_id,
395        } = self;
396        f.debug_struct("FileInfo")
397            .field("volume", volume)
398            .field("name", name)
399            .field("version_id", version_id)
400            .field("is_latest", is_latest)
401            .field("deleted", deleted)
402            .field("transition_status", transition_status)
403            .field("transitioned_objname", transitioned_objname)
404            .field("transition_tier", transition_tier)
405            .field("transition_version_id", transition_version_id)
406            .field("transition_version", transition_version)
407            .field("transition_version_state", transition_version_state)
408            .field("expire_restored", expire_restored)
409            .field("data_dir", data_dir)
410            .field("mod_time", mod_time)
411            .field("size", size)
412            .field("mode", mode)
413            .field("written_by_version", written_by_version)
414            .field("metadata", &RedactedMetadata(metadata))
415            .field("parts", parts)
416            .field("erasure", erasure)
417            .field("mark_deleted", mark_deleted)
418            .field("replication_state_internal", replication_state_internal)
419            .field("data", &ElidedBytes(data))
420            .field("num_versions", num_versions)
421            .field("successor_mod_time", successor_mod_time)
422            .field("fresh", fresh)
423            .field("idx", idx)
424            .field("checksum", &ElidedBytes(checksum))
425            .field("versioned", versioned)
426            .field("uses_legacy_checksum", uses_legacy_checksum)
427            .field("overwrite_tier_free_version_id", overwrite_tier_free_version_id)
428            .finish()
429    }
430}
431
432#[derive(Deserialize)]
433#[serde(remote = "FileInfo")]
434struct FileInfoMapDef {
435    volume: String,
436    name: String,
437    version_id: Option<Uuid>,
438    is_latest: bool,
439    deleted: bool,
440    transition_status: String,
441    transitioned_objname: String,
442    transition_tier: String,
443    transition_version_id: Option<Uuid>,
444    #[serde(default)]
445    transition_version: Option<String>,
446    #[serde(default)]
447    transition_version_state: TransitionVersionState,
448    expire_restored: bool,
449    data_dir: Option<Uuid>,
450    mod_time: Option<OffsetDateTime>,
451    size: i64,
452    mode: Option<u32>,
453    written_by_version: Option<u64>,
454    metadata: HashMap<String, String>,
455    parts: Vec<ObjectPartInfo>,
456    erasure: ErasureInfo,
457    mark_deleted: bool,
458    replication_state_internal: Option<ReplicationState>,
459    data: Option<Bytes>,
460    num_versions: usize,
461    successor_mod_time: Option<OffsetDateTime>,
462    fresh: bool,
463    idx: usize,
464    checksum: Option<Bytes>,
465    versioned: bool,
466    uses_legacy_checksum: bool,
467    #[serde(default)]
468    overwrite_tier_free_version_id: Option<Uuid>,
469}
470
471#[derive(Deserialize)]
472#[serde(untagged)]
473enum TransitionVersionOrExpireRestored {
474    TransitionVersion(Option<String>),
475    ExpireRestored(bool),
476}
477
478const FILE_INFO_FIELDS: &[&str] = &[
479    "volume",
480    "name",
481    "version_id",
482    "is_latest",
483    "deleted",
484    "transition_status",
485    "transitioned_objname",
486    "transition_tier",
487    "transition_version_id",
488    "transition_version",
489    "transition_version_state",
490    "expire_restored",
491    "data_dir",
492    "mod_time",
493    "size",
494    "mode",
495    "written_by_version",
496    "metadata",
497    "parts",
498    "erasure",
499    "mark_deleted",
500    "replication_state_internal",
501    "data",
502    "num_versions",
503    "successor_mod_time",
504    "fresh",
505    "idx",
506    "checksum",
507    "versioned",
508    "uses_legacy_checksum",
509    "overwrite_tier_free_version_id",
510];
511
512impl Serialize for FileInfo {
513    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
514    where
515        S: serde::Serializer,
516    {
517        let mut map = serializer
518            .serialize_map(Some(FILE_INFO_FIELDS.len() - usize::from(self.overwrite_tier_free_version_id.is_none())))?;
519        map.serialize_entry("volume", &self.volume)?;
520        map.serialize_entry("name", &self.name)?;
521        map.serialize_entry("version_id", &self.version_id)?;
522        map.serialize_entry("is_latest", &self.is_latest)?;
523        map.serialize_entry("deleted", &self.deleted)?;
524        map.serialize_entry("transition_status", &self.transition_status)?;
525        map.serialize_entry("transitioned_objname", &self.transitioned_objname)?;
526        map.serialize_entry("transition_tier", &self.transition_tier)?;
527        map.serialize_entry("transition_version_id", &self.transition_version_id)?;
528        map.serialize_entry("transition_version", &self.transition_version)?;
529        map.serialize_entry("transition_version_state", &self.transition_version_state)?;
530        map.serialize_entry("expire_restored", &self.expire_restored)?;
531        map.serialize_entry("data_dir", &self.data_dir)?;
532        map.serialize_entry("mod_time", &self.mod_time)?;
533        map.serialize_entry("size", &self.size)?;
534        map.serialize_entry("mode", &self.mode)?;
535        map.serialize_entry("written_by_version", &self.written_by_version)?;
536        map.serialize_entry("metadata", &self.metadata)?;
537        map.serialize_entry("parts", &self.parts)?;
538        map.serialize_entry("erasure", &self.erasure)?;
539        map.serialize_entry("mark_deleted", &self.mark_deleted)?;
540        map.serialize_entry("replication_state_internal", &self.replication_state_internal)?;
541        map.serialize_entry("data", &self.data)?;
542        map.serialize_entry("num_versions", &self.num_versions)?;
543        map.serialize_entry("successor_mod_time", &self.successor_mod_time)?;
544        map.serialize_entry("fresh", &self.fresh)?;
545        map.serialize_entry("idx", &self.idx)?;
546        map.serialize_entry("checksum", &self.checksum)?;
547        map.serialize_entry("versioned", &self.versioned)?;
548        map.serialize_entry("uses_legacy_checksum", &self.uses_legacy_checksum)?;
549        if let Some(id) = self.overwrite_tier_free_version_id {
550            map.serialize_entry("overwrite_tier_free_version_id", &id)?;
551        }
552        map.end()
553    }
554}
555
556impl<'de> Deserialize<'de> for FileInfo {
557    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
558    where
559        D: serde::Deserializer<'de>,
560    {
561        struct FileInfoVisitor;
562
563        impl<'de> Visitor<'de> for FileInfoVisitor {
564            type Value = FileInfo;
565
566            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
567                formatter.write_str("a FileInfo map or supported positional array")
568            }
569
570            fn visit_map<A>(self, map: A) -> std::result::Result<Self::Value, A::Error>
571            where
572                A: MapAccess<'de>,
573            {
574                FileInfoMapDef::deserialize(MapAccessDeserializer::new(map))
575            }
576
577            fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error>
578            where
579                A: SeqAccess<'de>,
580            {
581                // RUSTFS_COMPAT_TODO(rustfs-5509): beta.11 and beta.12 wrote incompatible positional arrays. Remove after every supported direct-upgrade release writes named maps and retained RPC payloads cannot contain either array.
582                let declared_len = seq.size_hint();
583                if let Some(len) = declared_len
584                    && len != 28
585                    && len != 30
586                {
587                    return Err(de::Error::invalid_length(len, &self));
588                }
589
590                macro_rules! next_field {
591                    ($field:literal) => {
592                        seq.next_element()?
593                            .ok_or_else(|| de::Error::missing_field($field))?
594                    };
595                }
596
597                let volume = next_field!("volume");
598                let name = next_field!("name");
599                let version_id = next_field!("version_id");
600                let is_latest = next_field!("is_latest");
601                let deleted = next_field!("deleted");
602                let transition_status = next_field!("transition_status");
603                let transitioned_objname = next_field!("transitioned_objname");
604                let transition_tier = next_field!("transition_tier");
605                let transition_version_id = next_field!("transition_version_id");
606                let transition_or_expire = next_field!("transition_version or expire_restored");
607                let (transition_version, transition_version_state, expire_restored) = match transition_or_expire {
608                    TransitionVersionOrExpireRestored::TransitionVersion(transition_version) => (
609                        transition_version,
610                        next_field!("transition_version_state"),
611                        next_field!("expire_restored"),
612                    ),
613                    TransitionVersionOrExpireRestored::ExpireRestored(expire_restored) => {
614                        (None, TransitionVersionState::Unknown, expire_restored)
615                    }
616                };
617                let data_dir = next_field!("data_dir");
618                let mod_time = next_field!("mod_time");
619                let size = next_field!("size");
620                let mode = next_field!("mode");
621                let written_by_version = next_field!("written_by_version");
622                let metadata = next_field!("metadata");
623                let parts = next_field!("parts");
624                let erasure = next_field!("erasure");
625                let mark_deleted = next_field!("mark_deleted");
626                let replication_state_internal = next_field!("replication_state_internal");
627                let data = next_field!("data");
628                let num_versions = next_field!("num_versions");
629                let successor_mod_time = next_field!("successor_mod_time");
630                let fresh = next_field!("fresh");
631                let idx = next_field!("idx");
632                let checksum = next_field!("checksum");
633                let versioned = next_field!("versioned");
634                let uses_legacy_checksum = next_field!("uses_legacy_checksum");
635
636                if seq.next_element::<de::IgnoredAny>()?.is_some() {
637                    return Err(de::Error::invalid_length(declared_len.unwrap_or(29), &self));
638                }
639
640                Ok(FileInfo {
641                    volume,
642                    name,
643                    version_id,
644                    is_latest,
645                    deleted,
646                    transition_status,
647                    transitioned_objname,
648                    transition_tier,
649                    transition_version_id,
650                    transition_version,
651                    transition_version_state,
652                    expire_restored,
653                    data_dir,
654                    mod_time,
655                    size,
656                    mode,
657                    written_by_version,
658                    metadata,
659                    parts,
660                    erasure,
661                    mark_deleted,
662                    replication_state_internal,
663                    data,
664                    num_versions,
665                    successor_mod_time,
666                    fresh,
667                    idx,
668                    checksum,
669                    versioned,
670                    uses_legacy_checksum,
671                    overwrite_tier_free_version_id: None,
672                })
673            }
674        }
675
676        let mut file = deserializer.deserialize_struct("FileInfo", FILE_INFO_FIELDS, FileInfoVisitor)?;
677        if !file.parts.is_empty() {
678            file.hydrate_shard_integrity().map_err(de::Error::custom)?;
679        }
680        Ok(file)
681    }
682}
683
684/// Selects the validation policy for a trusted operation boundary.
685///
686/// This mode is deliberately caller-selected and is never inferred from
687/// serialized [`FileInfo`] state such as `deleted` or `transition_status`.
688#[derive(Debug, Clone, Copy, PartialEq, Eq)]
689pub enum ValidationMode {
690    /// Validate the complete storage erasure layout before payload access.
691    RequireErasure,
692    /// Validate delete metadata without requiring a payload erasure layout.
693    DeleteOnly,
694}
695
696/// Erasure geometry that passed the storage-layout validation policy.
697#[derive(Debug, Clone, Copy, PartialEq, Eq)]
698pub struct ValidatedErasureLayout {
699    data_blocks: usize,
700    block_size: usize,
701    shard_size: usize,
702}
703
704impl ValidatedErasureLayout {
705    /// Calculates a shard file size without overflowing the metadata's integer format.
706    pub fn shard_file_size(&self, total_length: usize) -> Option<i64> {
707        let full_blocks = total_length / self.block_size;
708        let last_block_size = total_length % self.block_size;
709        let last_shard_size = checked_calc_shard_size(last_block_size, self.data_blocks)?;
710        let shard_file_size = full_blocks.checked_mul(self.shard_size)?.checked_add(last_shard_size)?;
711        i64::try_from(shard_file_size).ok()
712    }
713}
714
715fn mark_fileinfo_part(bitmap: &mut [u64; FILEINFO_PART_BITMAP_WORDS], part_number: usize) -> bool {
716    let Some(index) = part_number.checked_sub(1).filter(|&index| index < MAX_FILEINFO_PARTS) else {
717        return false;
718    };
719    let word = index / FILEINFO_PART_BITMAP_WORD_BITS;
720    let mask = 1u64 << (index % FILEINFO_PART_BITMAP_WORD_BITS);
721    let was_absent = bitmap[word] & mask == 0;
722    bitmap[word] |= mask;
723    was_absent
724}
725
726fn has_fileinfo_part(bitmap: &[u64; FILEINFO_PART_BITMAP_WORDS], part_number: usize) -> bool {
727    let Some(index) = part_number.checked_sub(1).filter(|&index| index < MAX_FILEINFO_PARTS) else {
728        return false;
729    };
730    bitmap[index / FILEINFO_PART_BITMAP_WORD_BITS] & (1u64 << (index % FILEINFO_PART_BITMAP_WORD_BITS)) != 0
731}
732
733/// Validates that an erasure `distribution` is a permutation of `1..=n`.
734///
735/// A well-formed distribution has exactly `n` entries and each 1-based slot
736/// index in `1..=n` appears exactly once. Corrupt or adversarial `xl.meta`
737/// can carry values of `0` or greater than `n`, which are later used as
738/// `distribution[k] - 1` indices into fixed-size vectors and would trigger a
739/// `usize` underflow / out-of-bounds panic. Rejecting such distributions here
740/// lets the metadata surface as a clean quorum/corruption error instead.
741pub(crate) fn is_valid_distribution(distribution: &[usize], n: usize) -> bool {
742    if n == 0 || n > MAX_ERASURE_SHARDS || distribution.len() != n {
743        return false;
744    }
745
746    let mut seen = [false; MAX_ERASURE_SHARDS];
747    for &block_idx in distribution {
748        // Valid 1-based slots are `1..=n`; anything else (including `0`) is invalid.
749        if block_idx < 1 || block_idx > n {
750            return false;
751        }
752        let slot = block_idx - 1;
753        if seen[slot] {
754            // Duplicate slot: not a permutation.
755            return false;
756        }
757        seen[slot] = true;
758    }
759
760    true
761}
762
763impl FileInfo {
764    pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
765        let indices = {
766            let cardinality = data_blocks + parity_blocks;
767            let mut nums = vec![0; cardinality];
768            let key_crc = {
769                let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
770                hasher.update(object.as_bytes());
771                hasher.finalize() as u32
772            };
773
774            let start = key_crc as usize % cardinality;
775            for i in 1..=cardinality {
776                nums[i - 1] = 1 + ((start + i) % cardinality);
777            }
778
779            nums
780        };
781        Self {
782            erasure: ErasureInfo {
783                algorithm: String::from(ERASURE_ALGORITHM),
784                data_blocks,
785                parity_blocks,
786                block_size: BLOCK_SIZE_V2,
787                distribution: indices,
788                ..Default::default()
789            },
790            ..Default::default()
791        }
792    }
793
794    fn validate_erasure_geometry_with_index(&self, index: usize) -> Result<ValidatedErasureLayout> {
795        let erasure = &self.erasure;
796        if erasure.data_blocks == 0 || erasure.block_size == 0 || erasure.data_blocks < erasure.parity_blocks {
797            return Err(Error::FileCorrupt);
798        }
799
800        let total_blocks = erasure
801            .data_blocks
802            .checked_add(erasure.parity_blocks)
803            .ok_or(Error::FileCorrupt)?;
804        if total_blocks > MAX_ERASURE_SHARDS
805            || index == 0
806            || index > total_blocks
807            || !is_valid_distribution(&erasure.distribution, total_blocks)
808        {
809            return Err(Error::FileCorrupt);
810        }
811
812        let shard_size = checked_calc_shard_size(erasure.block_size, erasure.data_blocks)
813            .filter(|&size| i64::try_from(size).is_ok())
814            .ok_or(Error::FileCorrupt)?;
815        i64::try_from(erasure.block_size).map_err(|_| Error::FileCorrupt)?;
816
817        let layout = ValidatedErasureLayout {
818            data_blocks: erasure.data_blocks,
819            block_size: erasure.block_size,
820            shard_size,
821        };
822        let object_size = usize::try_from(self.size).map_err(|_| Error::FileCorrupt)?;
823        layout.shard_file_size(object_size).ok_or(Error::FileCorrupt)?;
824        Ok(layout)
825    }
826
827    fn validate_erasure_geometry(&self) -> Result<ValidatedErasureLayout> {
828        self.validate_erasure_geometry_with_index(self.erasure.index)
829    }
830
831    fn validate_collection_bounds(&self) -> Result<()> {
832        if self.parts.len() > MAX_FILEINFO_PARTS
833            || self.erasure.checksums.len() > MAX_FILEINFO_CHECKSUMS
834            || self.erasure.distribution.len() > MAX_ERASURE_SHARDS
835        {
836            return Err(Error::FileCorrupt);
837        }
838        Ok(())
839    }
840
841    fn validate_collection_contents(&self, layout: Option<&ValidatedErasureLayout>) -> Result<()> {
842        if layout.is_some() && self.size > 0 && self.parts.is_empty() {
843            return Err(Error::FileCorrupt);
844        }
845
846        let mut part_numbers = [0u64; FILEINFO_PART_BITMAP_WORDS];
847        for part in &self.parts {
848            if let Some(layout) = layout {
849                i64::try_from(part.size).map_err(|_| Error::FileCorrupt)?;
850                layout.shard_file_size(part.size).ok_or(Error::FileCorrupt)?;
851
852                // A negative `actual_size` is the documented "unknown size" sentinel for
853                // compressed streaming objects (see `ObjectPartInfo::actual_size` /
854                // `ObjectInfo::get_actual_size`), written to xl.meta by both RustFS and
855                // MinIO. Only shard-validate a real, non-negative size; rejecting the
856                // sentinel would make legitimate compressed objects unreadable.
857                if let Ok(actual_size) = usize::try_from(part.actual_size) {
858                    layout.shard_file_size(actual_size).ok_or(Error::FileCorrupt)?;
859                }
860            }
861            if !mark_fileinfo_part(&mut part_numbers, part.number) {
862                return Err(Error::FileCorrupt);
863            }
864        }
865
866        let mut checksum_parts = [0u64; FILEINFO_PART_BITMAP_WORDS];
867        for checksum in &self.erasure.checksums {
868            if !has_fileinfo_part(&part_numbers, checksum.part_number)
869                || !mark_fileinfo_part(&mut checksum_parts, checksum.part_number)
870            {
871                return Err(Error::FileCorrupt);
872            }
873        }
874        Ok(())
875    }
876
877    fn validate_delete_marker_shape(&self, expected_index: usize, allow_nil_version_id: bool) -> Result<()> {
878        if !self.deleted {
879            return Err(Error::FileCorrupt);
880        }
881        let erasure = &self.erasure;
882        let stored_free_version = contains_key_str(&self.metadata, SUFFIX_FREE_VERSION);
883        if self.mod_time.is_none_or(|mod_time| mod_time <= OffsetDateTime::UNIX_EPOCH)
884            || (!allow_nil_version_id && self.version_id.is_some_and(|version_id| version_id.is_nil()))
885            || self.transition_version_id.is_some_and(|version_id| version_id.is_nil())
886            || self
887                .transition_version
888                .as_ref()
889                .is_some_and(|version_id| version_id.is_empty())
890            || self.size != 0
891            || self.data_dir.is_some()
892            || self.mode.is_some()
893            || self.written_by_version.is_some()
894            || self.data.is_some()
895            || self.checksum.is_some()
896            || !self.parts.is_empty()
897            || !erasure.algorithm.is_empty()
898            || erasure.data_blocks != 0
899            || erasure.parity_blocks != 0
900            || erasure.block_size != 0
901            || erasure.index != expected_index
902            || !erasure.distribution.is_empty()
903            || !erasure.checksums.is_empty()
904            || stored_free_version != self.tier_free_version()
905            || (stored_free_version && (self.transition_tier.is_empty() || self.transitioned_objname.is_empty()))
906        {
907            return Err(Error::FileCorrupt);
908        }
909        Ok(())
910    }
911
912    fn validate_tier_free_version_delete_shape(&self) -> Result<()> {
913        let erasure = &self.erasure;
914        if !self.deleted
915            || !self.tier_free_version()
916            || self.version_id.is_none_or(|version_id| version_id.is_nil())
917            || self.mod_time.is_some()
918            || self.is_latest
919            || !self.transition_status.is_empty()
920            || !self.transitioned_objname.is_empty()
921            || !self.transition_tier.is_empty()
922            || self.transition_version_id.is_some()
923            || self.transition_version.is_some()
924            || self.expire_restored
925            || self.size != 0
926            || self.data_dir.is_some()
927            || self.mode.is_some()
928            || self.written_by_version.is_some()
929            || self.mark_deleted
930            || self.replication_state_internal.is_some()
931            || self.data.is_some()
932            || self.num_versions != 0
933            || self.successor_mod_time.is_some()
934            || self.fresh
935            || self.idx != 0
936            || self.checksum.is_some()
937            || self.versioned
938            || self.uses_legacy_checksum
939            || !self.parts.is_empty()
940            || self.metadata.is_empty()
941            || self.metadata.len() > 2
942            || self
943                .metadata
944                .iter()
945                .any(|(key, value)| !has_internal_suffix(key, SUFFIX_TIER_FV_MARKER) || !value.is_empty())
946            || !erasure.algorithm.is_empty()
947            || erasure.data_blocks != 0
948            || erasure.parity_blocks != 0
949            || erasure.block_size != 0
950            || erasure.index != 0
951            || !erasure.distribution.is_empty()
952            || !erasure.checksums.is_empty()
953        {
954            return Err(Error::FileCorrupt);
955        }
956        Ok(())
957    }
958
959    /// Validates this metadata for the caller-selected operation boundary and
960    /// returns the payload erasure layout when one is established.
961    ///
962    /// All modes enforce collection bounds and checksum-to-part associations.
963    /// Only [`ValidationMode::RequireErasure`] establishes a payload erasure
964    /// layout; the other modes cannot be upgraded based on serialized flags and
965    /// return `None`.
966    pub fn validate(&self, mode: ValidationMode) -> Result<Option<ValidatedErasureLayout>> {
967        self.validate_collection_bounds()?;
968        if let (Some(version), Some(version_id)) = (&self.transition_version, self.transition_version_id)
969            && Uuid::parse_str(version).ok() != Some(version_id)
970        {
971            return Err(Error::FileCorrupt);
972        }
973        let transition_state_valid = match self.transition_version_state {
974            TransitionVersionState::Unknown => true,
975            TransitionVersionState::KnownDisabled => self.transition_version.is_none() && self.transition_version_id.is_none(),
976            TransitionVersionState::SuspendedNull => {
977                self.transition_version.as_deref() == Some("null") && self.transition_version_id.is_none()
978            }
979            TransitionVersionState::Exact => self
980                .transition_version
981                .as_deref()
982                .is_some_and(|version| version != "null" && !version.is_empty()),
983        };
984        if !transition_state_valid {
985            return Err(Error::FileCorrupt);
986        }
987
988        let erasure_layout = match mode {
989            ValidationMode::RequireErasure => Some(self.validate_erasure_geometry()?),
990            ValidationMode::DeleteOnly => {
991                self.validate_delete_marker_shape(0, false)?;
992                None
993            }
994        };
995        self.validate_collection_contents(erasure_layout.as_ref())?;
996
997        Ok(erasure_layout)
998    }
999
1000    /// Validate metadata returned by a disk or peer as a strict tagged union.
1001    /// Payload entries require complete erasure geometry, including
1002    /// purge-pending object versions whose replication state sets `deleted`.
1003    /// Only entries that are not valid payloads may fall back to the canonical
1004    /// delete-marker shape, so `deleted` cannot bypass payload validation.
1005    pub fn validate_for_metadata_read(&self) -> Result<()> {
1006        if self.validate(ValidationMode::RequireErasure).is_ok() {
1007            return Ok(());
1008        }
1009
1010        self.validate(ValidationMode::DeleteOnly).map(|_| ())
1011    }
1012
1013    /// Cheap shape check for metadata that already passed
1014    /// [`Self::validate_for_metadata_read`] at its decode boundary.
1015    pub fn has_valid_metadata_shape(&self) -> bool {
1016        self.has_valid_erasure_geometry() || self.is_canonical_delete_marker()
1017    }
1018
1019    /// Returns whether this is a canonical delete marker rather than an
1020    /// erasure-backed object version in a purge-pending state.
1021    pub fn is_canonical_delete_marker(&self) -> bool {
1022        self.validate_delete_marker_shape(0, false).is_ok()
1023    }
1024
1025    /// Storage-only marker classification after an absent version ID has been
1026    /// normalized to the internal nil UUID. Network and disk boundaries must
1027    /// continue to use [`Self::is_canonical_delete_marker`].
1028    pub(crate) fn is_storage_delete_marker(&self) -> bool {
1029        self.validate_delete_marker_shape(0, true).is_ok()
1030    }
1031
1032    /// Whether this is a canonical delete marker carrying only the per-disk
1033    /// index assigned by an older rename coordinator.
1034    pub fn is_legacy_indexed_delete_marker(&self) -> bool {
1035        self.erasure.index > 0
1036            && self.erasure.index <= MAX_ERASURE_SHARDS
1037            && self.validate_delete_marker_shape(self.erasure.index, false).is_ok()
1038    }
1039
1040    /// Validates complete metadata before a write fanout assigns its per-disk
1041    /// shard index. An index of zero is treated only as the pending assignment;
1042    /// every other geometry and collection invariant remains mandatory.
1043    pub fn validate_for_erasure_write(&self) -> Result<()> {
1044        self.validate_collection_bounds()?;
1045        let index = if self.erasure.index == 0 { 1 } else { self.erasure.index };
1046        let layout = self.validate_erasure_geometry_with_index(index)?;
1047        self.validate_collection_contents(Some(&layout))
1048    }
1049
1050    /// Validates metadata used to mutate a version list during a delete.
1051    /// Delete-marker creation requires the canonical marker shape; other
1052    /// delete operations do not access payload shards but still validate all
1053    /// collection bounds and checksum-to-part associations before mutation.
1054    pub fn validate_for_delete_operation(&self) -> Result<()> {
1055        self.validate_collection_bounds()?;
1056
1057        if self.deleted && self.tier_free_version() {
1058            return self.validate_tier_free_version_delete_shape();
1059        }
1060
1061        if self.deleted {
1062            return self.validate(ValidationMode::DeleteOnly).map(|_| ());
1063        }
1064
1065        self.validate_collection_contents(None)
1066    }
1067
1068    /// Performs the cheap layout check used by quorum selection after a decode
1069    /// boundary has already called [`Self::validate`].
1070    pub fn has_valid_erasure_geometry(&self) -> bool {
1071        self.validate_erasure_geometry().is_ok()
1072    }
1073
1074    /// Validates the complete payload metadata shape.
1075    ///
1076    /// Repeated quorum passes over metadata that was already fully validated
1077    /// should use [`Self::has_valid_erasure_geometry`] instead.
1078    pub fn is_valid(&self) -> bool {
1079        self.validate(ValidationMode::RequireErasure).is_ok()
1080    }
1081
1082    pub fn get_etag(&self) -> Option<String> {
1083        self.metadata.get("etag").cloned()
1084    }
1085
1086    pub fn write_quorum(&self, quorum: usize) -> usize {
1087        if self.deleted && !self.has_valid_erasure_geometry() {
1088            return quorum;
1089        }
1090
1091        if self.erasure.data_blocks == self.erasure.parity_blocks {
1092            return self.erasure.data_blocks + 1;
1093        }
1094
1095        self.erasure.data_blocks
1096    }
1097
1098    pub fn marshal_msg(&self) -> Result<Vec<u8>> {
1099        let mut buf = Vec::new();
1100
1101        self.serialize(&mut Serializer::new(&mut buf))?;
1102
1103        Ok(buf)
1104    }
1105
1106    pub fn unmarshal(buf: &[u8]) -> Result<Self> {
1107        let t: FileInfo = rmp_serde::from_slice(buf)?;
1108        Ok(t)
1109    }
1110
1111    #[allow(clippy::too_many_arguments)]
1112    pub fn add_object_part(
1113        &mut self,
1114        num: usize,
1115        etag: String,
1116        part_size: usize,
1117        mod_time: Option<OffsetDateTime>,
1118        actual_size: i64,
1119        index: Option<Bytes>,
1120        checksums: Option<HashMap<String, String>>,
1121    ) {
1122        let part = ObjectPartInfo {
1123            etag,
1124            number: num,
1125            size: part_size,
1126            mod_time,
1127            actual_size,
1128            index,
1129            checksums,
1130            error: None,
1131            integrity: None,
1132        };
1133
1134        for p in self.parts.iter_mut() {
1135            if p.number == num {
1136                *p = part;
1137                return;
1138            }
1139        }
1140
1141        self.parts.push(part);
1142
1143        self.parts.sort_by_key(|a| a.number);
1144    }
1145
1146    // to_part_offset gets the part index where offset is located, returns part index and offset
1147    pub fn to_part_offset(&self, offset: usize) -> Result<(usize, usize)> {
1148        if self.size > 0 && self.parts.is_empty() {
1149            return Err(Error::FileCorrupt);
1150        }
1151
1152        if offset == 0 {
1153            return Ok((0, 0));
1154        }
1155
1156        let mut part_offset = offset;
1157        for (i, part) in self.parts.iter().enumerate() {
1158            let part_index = i;
1159            if part_offset < part.size {
1160                return Ok((part_index, part_offset));
1161            }
1162
1163            part_offset -= part.size
1164        }
1165
1166        Err(Error::other("part not found"))
1167    }
1168
1169    pub fn set_healing(&mut self) {
1170        insert_str(&mut self.metadata, SUFFIX_HEALING, "true".to_string());
1171    }
1172
1173    /// Reader for the marker [`Self::set_healing`] writes: true when this
1174    /// FileInfo is being committed by the heal path.
1175    pub fn is_healing(&self) -> bool {
1176        contains_key_str(&self.metadata, SUFFIX_HEALING)
1177    }
1178
1179    pub fn set_tier_free_version_id(&mut self, version_id: &str) {
1180        insert_str(&mut self.metadata, SUFFIX_TIER_FV_ID, version_id.to_string());
1181    }
1182
1183    pub fn tier_free_version_id(&self) -> String {
1184        get_str(&self.metadata, SUFFIX_TIER_FV_ID).unwrap_or_default()
1185    }
1186
1187    pub fn set_tier_free_version(&mut self) {
1188        insert_str(&mut self.metadata, SUFFIX_TIER_FV_MARKER, "".to_string());
1189    }
1190
1191    pub fn set_skip_tier_free_version(&mut self) {
1192        insert_str(&mut self.metadata, SUFFIX_TIER_SKIP_FV_ID, "".to_string());
1193    }
1194
1195    pub fn skip_tier_free_version(&self) -> bool {
1196        contains_key_str(&self.metadata, SUFFIX_TIER_SKIP_FV_ID)
1197    }
1198
1199    pub fn tier_free_version(&self) -> bool {
1200        contains_key_str(&self.metadata, SUFFIX_TIER_FV_MARKER)
1201    }
1202
1203    pub fn set_inline_data(&mut self) {
1204        insert_str(&mut self.metadata, SUFFIX_INLINE_DATA, "true".to_string());
1205    }
1206
1207    pub fn set_data_moved(&mut self) {
1208        let tags_proof = format!("v1:{}", self.metadata.get(AMZ_OBJECT_TAGGING).map(String::as_str).unwrap_or_default());
1209        insert_str(&mut self.metadata, SUFFIX_DATA_MOVED_TAGS, tags_proof);
1210        insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, "true".to_string());
1211    }
1212
1213    pub fn acknowledge_data_movement(&mut self) {
1214        // Keep both empty aliases so mixed-version disks retain one metadata identity.
1215        insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, String::new());
1216    }
1217
1218    pub fn set_object_transaction_epoch(&mut self, epoch: Uuid) {
1219        insert_str(&mut self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH, epoch.to_string());
1220    }
1221
1222    /// Bind a newly created delete marker to the destination bucket generation.
1223    pub fn set_delete_marker_incarnation(&mut self, incarnation: Uuid) {
1224        if self.deleted && !incarnation.is_nil() {
1225            insert_str(
1226                &mut self.metadata,
1227                rustfs_utils::http::metadata_compat::SUFFIX_BUCKET_INCARNATION_ID,
1228                incarnation.to_string(),
1229            );
1230        }
1231    }
1232
1233    /// Legacy, malformed, or conflicting stamps cannot authorize marker cleanup.
1234    pub fn delete_marker_incarnation(&self) -> Option<Uuid> {
1235        if !self.deleted || self.tier_free_version() {
1236            return None;
1237        }
1238        get_consistent_str(&self.metadata, rustfs_utils::http::metadata_compat::SUFFIX_BUCKET_INCARNATION_ID)
1239            .and_then(|value| Uuid::parse_str(value).ok())
1240            .filter(|id| !id.is_nil())
1241    }
1242
1243    pub fn object_transaction_epoch(&self) -> Result<Option<Uuid>> {
1244        if !contains_key_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH) {
1245            return Ok(None);
1246        }
1247        let value = get_consistent_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH).ok_or(Error::FileCorrupt)?;
1248        let epoch = Uuid::parse_str(value).map_err(|_| Error::FileCorrupt)?;
1249        if epoch.is_nil() {
1250            return Err(Error::FileCorrupt);
1251        }
1252        Ok(Some(epoch))
1253    }
1254
1255    pub fn inline_data(&self) -> bool {
1256        contains_key_str(&self.metadata, SUFFIX_INLINE_DATA) && !self.is_remote()
1257    }
1258
1259    /// Check if the object is compressed
1260    pub fn is_compressed(&self) -> bool {
1261        contains_key_str(&self.metadata, SUFFIX_COMPRESSION)
1262    }
1263
1264    /// Check if the object is remote (transitioned to another tier)
1265    pub fn is_remote(&self) -> bool {
1266        if self.transition_status != TRANSITION_COMPLETE {
1267            return false;
1268        }
1269        !is_restored_object_on_disk(&self.metadata)
1270    }
1271
1272    /// Get the data directory for this object
1273    pub fn get_data_dir(&self) -> String {
1274        if self.deleted {
1275            return "delete-marker".to_string();
1276        }
1277        self.data_dir.map_or_else(|| "".to_string(), |dir| dir.to_string())
1278    }
1279
1280    /// Read quorum returns expected read quorum for this FileInfo
1281    pub fn read_quorum(&self, dquorum: usize) -> usize {
1282        if self.deleted {
1283            return dquorum;
1284        }
1285        self.erasure.data_blocks
1286    }
1287
1288    /// Create a shallow copy with minimal information for READ MRF checks
1289    pub fn shallow_copy(&self) -> Self {
1290        Self {
1291            volume: self.volume.clone(),
1292            name: self.name.clone(),
1293            version_id: self.version_id,
1294            deleted: self.deleted,
1295            erasure: self.erasure.clone(),
1296            ..Default::default()
1297        }
1298    }
1299
1300    /// Check if this FileInfo equals another FileInfo
1301    pub fn equals(&self, other: &FileInfo) -> bool {
1302        // Check if both are compressed or both are not compressed
1303        if self.is_compressed() != other.is_compressed() {
1304            tracing::warn!("equals: is_compressed is not equal, object_name={}", self.name);
1305            return false;
1306        }
1307
1308        // Check transition info
1309        if !self.transition_info_equals(other) {
1310            tracing::warn!("equals: transition_info_equals is not equal, object_name={}", self.name);
1311            return false;
1312        }
1313
1314        // Check mod time
1315        if self.mod_time != other.mod_time {
1316            tracing::warn!("equals: mod_time is not equal, object_name={}", self.name);
1317            return false;
1318        }
1319
1320        // Check erasure info
1321        if !self.erasure.equals(&other.erasure) {
1322            tracing::warn!("equals: erasure is not equal, object_name={}", self.name);
1323            return false;
1324        }
1325        true
1326    }
1327
1328    /// Check if transition related information are equal
1329    pub fn transition_info_equals(&self, other: &FileInfo) -> bool {
1330        self.transition_status == other.transition_status
1331            && self.transition_tier == other.transition_tier
1332            && self.transitioned_objname == other.transitioned_objname
1333            && self.transition_version_id == other.transition_version_id
1334            && self.transition_version == other.transition_version
1335            && self.transition_version_state == other.transition_version_state
1336    }
1337
1338    /// Check if metadata maps are equal
1339    pub fn metadata_equals(&self, other: &FileInfo) -> bool {
1340        if self.metadata.len() != other.metadata.len() {
1341            return false;
1342        }
1343        for (k, v) in &self.metadata {
1344            if other.metadata.get(k) != Some(v) {
1345                return false;
1346            }
1347        }
1348        true
1349    }
1350
1351    /// Check if replication related fields are equal
1352    pub fn replication_info_equals(&self, other: &FileInfo) -> bool {
1353        self.mark_deleted == other.mark_deleted && self.replication_state_internal == other.replication_state_internal
1354    }
1355
1356    pub fn version_purge_status(&self) -> VersionPurgeStatusType {
1357        self.replication_state_internal
1358            .as_ref()
1359            .map(|v| v.composite_version_purge_status())
1360            .unwrap_or(VersionPurgeStatusType::Empty)
1361    }
1362    pub fn replication_status(&self) -> ReplicationStatusType {
1363        self.replication_state_internal
1364            .as_ref()
1365            .map(|v| v.composite_replication_status())
1366            .unwrap_or(ReplicationStatusType::Empty)
1367    }
1368    pub fn delete_marker_replication_status(&self) -> ReplicationStatusType {
1369        if self.deleted {
1370            self.replication_state_internal
1371                .as_ref()
1372                .map(|v| v.composite_replication_status())
1373                .unwrap_or(ReplicationStatusType::Empty)
1374        } else {
1375            ReplicationStatusType::Empty
1376        }
1377    }
1378
1379    pub fn shard_file_size(&self, total_length: i64) -> i64 {
1380        self.erasure.shard_file_size(total_length)
1381    }
1382}
1383
1384#[derive(Debug, Default, Clone, Serialize, Deserialize)]
1385pub struct FileInfoVersions {
1386    // Name of the volume.
1387    pub volume: String,
1388
1389    // Name of the file.
1390    pub name: String,
1391
1392    // Represents the latest mod time of the
1393    // latest version.
1394    pub latest_mod_time: Option<OffsetDateTime>,
1395
1396    pub versions: Vec<FileInfo>,
1397    pub free_versions: Vec<FileInfo>,
1398}
1399
1400impl FileInfoVersions {
1401    pub fn find_version_index(&self, vid: Uuid) -> Option<usize> {
1402        self.versions.iter().position(|v| v.version_id == Some(vid))
1403    }
1404
1405    /// Calculate the total size of all versions for this object
1406    pub fn size(&self) -> i64 {
1407        self.versions.iter().map(|v| v.size).sum()
1408    }
1409}
1410
1411#[derive(Default, Serialize, Deserialize)]
1412pub struct RawFileInfo {
1413    pub buf: Vec<u8>,
1414}
1415
1416#[derive(Debug, Default, Clone, Serialize, Deserialize)]
1417pub struct FilesInfo {
1418    pub files: Vec<FileInfo>,
1419    pub is_truncated: bool,
1420}
1421
1422/// Parsed restore status of a restored object: the value persisted under
1423/// [`metadata_keys::RESTORE`]. filemeta owns this type so the persisted format
1424/// does not depend on an HTTP library DTO (backlog#1735 A3c); the field names
1425/// match the former `s3s::dto::RestoreStatus` so the rendered bytes are unchanged.
1426#[derive(Debug, Clone, Default, PartialEq, Eq)]
1427pub struct RestoreStatus {
1428    /// `ongoing-request`; `None` reads as not in progress.
1429    pub is_restore_in_progress: Option<bool>,
1430    /// `expiry-date`; required to render a finished restore.
1431    pub restore_expiry_date: Option<OffsetDateTime>,
1432}
1433
1434pub trait RestoreStatusOps {
1435    fn expiry(&self) -> Option<OffsetDateTime>;
1436    fn on_going(&self) -> bool;
1437    fn on_disk(&self) -> bool;
1438    fn to_string(&self) -> String;
1439    fn to_string2(&self) -> String;
1440}
1441
1442impl RestoreStatusOps for RestoreStatus {
1443    fn expiry(&self) -> Option<OffsetDateTime> {
1444        if self.on_going() {
1445            return None;
1446        }
1447        self.restore_expiry_date
1448    }
1449
1450    fn on_going(&self) -> bool {
1451        if let Some(on_going) = self.is_restore_in_progress {
1452            return on_going;
1453        }
1454        false
1455    }
1456
1457    fn on_disk(&self) -> bool {
1458        let expiry = self.expiry();
1459        if let Some(expiry0) = expiry
1460            && OffsetDateTime::now_utc().unix_timestamp() < expiry0.unix_timestamp()
1461        {
1462            return true;
1463        }
1464        false
1465    }
1466
1467    fn to_string(&self) -> String {
1468        if self.on_going() {
1469            return "ongoing-request=\"true\"".to_string();
1470        }
1471        format!(
1472            "ongoing-request=\"false\", expiry-date=\"{}\"",
1473            self.restore_expiry_date.unwrap().format(&Rfc3339).unwrap()
1474        )
1475    }
1476
1477    fn to_string2(&self) -> String {
1478        if self.on_going() {
1479            return "ongoing-request=\"true\"".to_string();
1480        }
1481        format!(
1482            "ongoing-request=\"false\", expiry-date=\"{}\"",
1483            self.restore_expiry_date.unwrap().format(&RFC1123).unwrap()
1484        )
1485    }
1486}
1487
1488/// Parse the `expiry-date` value carried by a persisted `x-amz-restore` header.
1489///
1490/// RustFS has always serialized this field as RFC3339, while MinIO (and any
1491/// object migrated in place from a MinIO deployment) writes Go's
1492/// `http.TimeFormat` — RFC1123 with a literal `GMT` zone. Accept both shapes
1493/// so a migrated "restored" object stays readable.
1494///
1495/// This is not only a HEAD 500: `is_restored_object_on_disk` fails open when
1496/// the header cannot be parsed, and `MetaObject::uses_data_dir` falls back to
1497/// it, so an unparseable restore header makes a live restored data dir look
1498/// unused (backlog#1342).
1499///
1500/// The write side deliberately stays on RFC3339 for now: an older node only
1501/// parses RFC3339, so switching the persisted format before this parser is
1502/// deployed everywhere would break rolling upgrades and rollbacks in the
1503/// opposite direction.
1504fn parse_restore_expiry_date(value: &str) -> Result<OffsetDateTime> {
1505    if let Ok(expiry) = OffsetDateTime::parse(value, &Rfc3339) {
1506        return Ok(expiry);
1507    }
1508    PrimitiveDateTime::parse(value, &RFC1123)
1509        .map(PrimitiveDateTime::assume_utc)
1510        .map_err(|_| Error::other(ERR_RESTORE_HDR_MALFORMED))
1511}
1512
1513pub fn parse_restore_obj_status(restore_hdr: &str) -> Result<RestoreStatus> {
1514    let tokens: Vec<&str> = restore_hdr.splitn(2, ",").collect();
1515    let progress_tokens: Vec<&str> = tokens[0].splitn(2, "=").collect();
1516    if progress_tokens.len() != 2 {
1517        return Err(Error::other(ERR_RESTORE_HDR_MALFORMED));
1518    }
1519    if progress_tokens[0].trim() != "ongoing-request" {
1520        return Err(Error::other(ERR_RESTORE_HDR_MALFORMED));
1521    }
1522
1523    match progress_tokens[1] {
1524        "true" | "\"true\"" if tokens.len() == 1 => {
1525            return Ok(RestoreStatus {
1526                is_restore_in_progress: Some(true),
1527                ..Default::default()
1528            });
1529        }
1530        "false" | "\"false\"" => {
1531            if tokens.len() != 2 {
1532                return Err(Error::other(ERR_RESTORE_HDR_MALFORMED));
1533            }
1534            let expiry_tokens: Vec<&str> = tokens[1].splitn(2, "=").collect();
1535            if expiry_tokens.len() != 2 {
1536                return Err(Error::other(ERR_RESTORE_HDR_MALFORMED));
1537            }
1538            if expiry_tokens[0].trim() != "expiry-date" {
1539                return Err(Error::other(ERR_RESTORE_HDR_MALFORMED));
1540            }
1541            let expiry = parse_restore_expiry_date(expiry_tokens[1].trim_matches('"'))?;
1542            return Ok(RestoreStatus {
1543                is_restore_in_progress: Some(false),
1544                restore_expiry_date: Some(expiry),
1545            });
1546        }
1547        _ => (),
1548    }
1549    Err(Error::other(ERR_RESTORE_HDR_MALFORMED))
1550}
1551
1552pub fn is_restored_object_on_disk(meta: &HashMap<String, String>) -> bool {
1553    if let Some(restore_hdr) = meta.get(metadata_keys::RESTORE)
1554        && let Ok(restore_status) = parse_restore_obj_status(restore_hdr)
1555    {
1556        return restore_status.on_disk();
1557    }
1558    false
1559}
1560
1561#[cfg(test)]
1562mod tests {
1563    use super::*;
1564    use proptest::collection::{hash_map, vec};
1565    use proptest::prelude::*;
1566    use time::macros::datetime;
1567
1568    // backlog#959 / ECA-18: the interleaved per-block bitrot subsystem in
1569    // rustfs-ecstore (BitrotWriter / bitrot_verify / bitrot_shard_file_size) is
1570    // only self-consistent for the streaming Highway variants. That safety rests
1571    // on production always resolving a streaming checksum algorithm, so pin the
1572    // default here: a part with no explicit ChecksumInfo must resolve to
1573    // HighwayHash256S. If this default ever changes, the ecstore bitrot layout
1574    // assumptions must be revisited in lockstep.
1575    #[test]
1576    fn get_checksum_info_defaults_to_highwayhash256s() {
1577        let ei = ErasureInfo::default();
1578        assert!(ei.checksums.is_empty(), "default ErasureInfo carries no checksums");
1579        let info = ei.get_checksum_info(1);
1580        assert_eq!(
1581            info.algorithm,
1582            HashAlgorithm::HighwayHash256S,
1583            "missing ChecksumInfo must default to the streaming HighwayHash256S algorithm"
1584        );
1585
1586        // A present entry is returned as-is; the default only applies on miss.
1587        let ei = ErasureInfo {
1588            checksums: vec![ChecksumInfo {
1589                part_number: 2,
1590                algorithm: HashAlgorithm::SHA256,
1591                hash: Bytes::new(),
1592            }],
1593            ..Default::default()
1594        };
1595        assert_eq!(ei.get_checksum_info(2).algorithm, HashAlgorithm::SHA256);
1596        // A part_number with no entry still falls back to the streaming default.
1597        assert_eq!(ei.get_checksum_info(99).algorithm, HashAlgorithm::HighwayHash256S);
1598    }
1599
1600    #[test]
1601    fn object_transaction_epoch_uses_consistent_dual_internal_metadata() {
1602        let mut fi = validation_test_fileinfo();
1603        assert_eq!(fi.object_transaction_epoch().expect("absent epoch should decode"), None);
1604
1605        let epoch = Uuid::new_v4();
1606        let epoch_text = epoch.to_string();
1607        fi.set_object_transaction_epoch(epoch);
1608        assert_eq!(fi.object_transaction_epoch().expect("written epoch should decode"), Some(epoch));
1609        assert_eq!(fi.metadata.get("x-rustfs-internal-object-transaction-epoch"), Some(&epoch_text));
1610        assert_eq!(fi.metadata.get("x-minio-internal-object-transaction-epoch"), Some(&epoch_text));
1611
1612        let mut rustfs_only = validation_test_fileinfo();
1613        rustfs_only
1614            .metadata
1615            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), epoch_text);
1616        assert_eq!(
1617            rustfs_only
1618                .object_transaction_epoch()
1619                .expect("single compatibility key should decode"),
1620            Some(epoch)
1621        );
1622
1623        let mut conflicting = fi.clone();
1624        conflicting
1625            .metadata
1626            .insert("x-minio-internal-object-transaction-epoch".to_string(), Uuid::new_v4().to_string());
1627        assert_eq!(conflicting.object_transaction_epoch(), Err(Error::FileCorrupt));
1628
1629        let mut malformed = validation_test_fileinfo();
1630        malformed
1631            .metadata
1632            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), "not-a-uuid".to_string());
1633        assert_eq!(malformed.object_transaction_epoch(), Err(Error::FileCorrupt));
1634
1635        let mut nil = validation_test_fileinfo();
1636        nil.set_object_transaction_epoch(Uuid::nil());
1637        assert_eq!(nil.object_transaction_epoch(), Err(Error::FileCorrupt));
1638    }
1639
1640    // backlog#949: distribution range/permutation validation.
1641    #[test]
1642    fn is_valid_distribution_accepts_permutation() {
1643        assert!(is_valid_distribution(&[1, 2, 3, 4], 4));
1644        assert!(is_valid_distribution(&[4, 2, 1, 3], 4));
1645        assert!(is_valid_distribution(&[1], 1));
1646    }
1647
1648    #[test]
1649    fn is_valid_distribution_rejects_zero_value() {
1650        // A `0` would underflow `block_idx - 1` in the shuffle helpers.
1651        assert!(!is_valid_distribution(&[0, 2, 3, 4], 4));
1652    }
1653
1654    #[test]
1655    fn is_valid_distribution_rejects_out_of_range_value() {
1656        // A value greater than N would index out of bounds in the shuffle helpers.
1657        assert!(!is_valid_distribution(&[1, 2, 3, 5], 4));
1658        assert!(!is_valid_distribution(&[usize::MAX, 2, 3, 4], 4));
1659    }
1660
1661    #[test]
1662    fn is_valid_distribution_rejects_duplicates() {
1663        // In-range but not a permutation.
1664        assert!(!is_valid_distribution(&[1, 1, 3, 4], 4));
1665    }
1666
1667    #[test]
1668    fn is_valid_distribution_rejects_wrong_length_and_empty() {
1669        assert!(!is_valid_distribution(&[1, 2, 3], 4));
1670        assert!(!is_valid_distribution(&[1, 2, 3, 4, 4], 4));
1671        assert!(!is_valid_distribution(&[], 0));
1672        assert!(!is_valid_distribution(&[], 4));
1673    }
1674
1675    fn distribution_test_fileinfo() -> FileInfo {
1676        // data=2, parity=2 => N=4, distribution is a valid permutation of 1..=4.
1677        let mut fi = FileInfo::new("bucket/object", 2, 2);
1678        fi.erasure.index = 1;
1679        fi
1680    }
1681
1682    #[test]
1683    fn is_valid_accepts_well_formed_distribution() {
1684        assert!(distribution_test_fileinfo().is_valid());
1685    }
1686
1687    #[test]
1688    fn is_valid_rejects_corrupt_distribution_values() {
1689        // Zero value.
1690        let mut fi = distribution_test_fileinfo();
1691        fi.erasure.distribution = vec![0, 2, 3, 4];
1692        assert!(!fi.is_valid());
1693
1694        // Out-of-range value.
1695        let mut fi = distribution_test_fileinfo();
1696        fi.erasure.distribution = vec![1, 2, 3, 9];
1697        assert!(!fi.is_valid());
1698
1699        // Duplicate value (not a permutation).
1700        let mut fi = distribution_test_fileinfo();
1701        fi.erasure.distribution = vec![1, 1, 3, 4];
1702        assert!(!fi.is_valid());
1703
1704        // Wrong length.
1705        let mut fi = distribution_test_fileinfo();
1706        fi.erasure.distribution = vec![1, 2, 3];
1707        assert!(!fi.is_valid());
1708    }
1709
1710    fn validation_test_fileinfo() -> FileInfo {
1711        let mut fi = FileInfo::new("bucket/object", 4, 2);
1712        fi.erasure.index = 1;
1713        fi.parts = vec![
1714            ObjectPartInfo {
1715                number: 1,
1716                ..Default::default()
1717            },
1718            ObjectPartInfo {
1719                number: 2,
1720                ..Default::default()
1721            },
1722        ];
1723        fi.erasure.checksums = vec![
1724            ChecksumInfo {
1725                part_number: 1,
1726                algorithm: HashAlgorithm::HighwayHash256S,
1727                hash: Bytes::from_static(b"checksum-one"),
1728            },
1729            ChecksumInfo {
1730                part_number: 2,
1731                algorithm: HashAlgorithm::SHA256,
1732                hash: Bytes::from_static(b"checksum-two"),
1733            },
1734        ];
1735        fi
1736    }
1737
1738    fn assert_file_corrupt(fi: &FileInfo, mode: ValidationMode) {
1739        assert_eq!(fi.validate(mode).expect_err("invalid FileInfo must be rejected"), Error::FileCorrupt);
1740    }
1741
1742    #[test]
1743    fn validate_require_erasure_returns_checked_layout() {
1744        let fi = validation_test_fileinfo();
1745        let layout = fi
1746            .validate(ValidationMode::RequireErasure)
1747            .expect("well-formed erasure metadata must validate")
1748            .expect("strict validation must return an erasure layout");
1749        assert_eq!(layout.data_blocks, 4);
1750        assert_eq!(layout.block_size, BLOCK_SIZE_V2);
1751        assert_eq!(layout.shard_size, calc_shard_size(BLOCK_SIZE_V2, 4));
1752        let shard_size = i64::try_from(layout.shard_size).expect("validated shard size must fit i64");
1753        assert_eq!(layout.shard_file_size(0), Some(0));
1754        assert_eq!(layout.shard_file_size(BLOCK_SIZE_V2 - 1), Some(shard_size));
1755        assert_eq!(layout.shard_file_size(BLOCK_SIZE_V2), Some(shard_size));
1756        assert_eq!(layout.shard_file_size(BLOCK_SIZE_V2 + 1), Some(shard_size + 2));
1757    }
1758
1759    #[test]
1760    fn validate_require_erasure_accepts_zero_parity() {
1761        let mut fi = FileInfo::new("bucket/object", 16, 0);
1762        fi.erasure.index = 1;
1763
1764        let layout = fi
1765            .validate(ValidationMode::RequireErasure)
1766            .expect("zero parity is a valid storage layout");
1767        assert_eq!(layout.expect("strict layout must be present").data_blocks, 16);
1768    }
1769
1770    #[test]
1771    fn validate_require_erasure_rejects_positive_size_without_parts() {
1772        let mut fi = FileInfo::new("bucket/object", 4, 2);
1773        fi.erasure.index = 1;
1774        fi.size = 1;
1775
1776        assert_eq!(fi.validate_for_metadata_read(), Err(Error::FileCorrupt));
1777        assert_eq!(fi.to_part_offset(0), Err(Error::FileCorrupt));
1778
1779        fi.size = 0;
1780        fi.validate_for_metadata_read()
1781            .expect("zero-byte object metadata may omit parts");
1782        assert_eq!(fi.to_part_offset(0), Ok((0, 0)));
1783    }
1784
1785    #[test]
1786    fn validate_for_erasure_write_only_relaxes_pending_shard_index() {
1787        let mut fi = validation_test_fileinfo();
1788        fi.erasure.index = 0;
1789
1790        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1791        fi.validate_for_erasure_write()
1792            .expect("write preparation must accept an index that fanout has not assigned yet");
1793
1794        fi.erasure.index = fi.erasure.data_blocks + fi.erasure.parity_blocks + 1;
1795        assert_eq!(fi.validate_for_erasure_write(), Err(Error::FileCorrupt));
1796
1797        fi.erasure.index = 0;
1798        fi.parts.push(fi.parts[0].clone());
1799        assert_eq!(fi.validate_for_erasure_write(), Err(Error::FileCorrupt));
1800    }
1801
1802    #[test]
1803    fn validate_require_erasure_rejects_invalid_geometry() {
1804        let mut fi = validation_test_fileinfo();
1805        fi.erasure.data_blocks = 0;
1806        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1807
1808        let mut fi = validation_test_fileinfo();
1809        fi.erasure.block_size = 0;
1810        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1811
1812        let mut fi = validation_test_fileinfo();
1813        fi.erasure.data_blocks = usize::MAX;
1814        fi.erasure.parity_blocks = 1;
1815        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1816
1817        let mut fi = validation_test_fileinfo();
1818        fi.erasure.data_blocks = 9;
1819        fi.erasure.parity_blocks = 8;
1820        fi.erasure.index = 1;
1821        fi.erasure.distribution = (1..=17).collect();
1822        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1823
1824        let mut fi = validation_test_fileinfo();
1825        fi.erasure.data_blocks = 1;
1826        fi.erasure.parity_blocks = 2;
1827        fi.erasure.index = 1;
1828        fi.erasure.distribution = vec![1, 2, 3];
1829        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1830
1831        let mut fi = validation_test_fileinfo();
1832        fi.erasure.index = 0;
1833        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1834
1835        let mut fi = validation_test_fileinfo();
1836        fi.erasure.index = 7;
1837        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1838
1839        let mut fi = validation_test_fileinfo();
1840        fi.erasure.distribution = vec![1, 2, 3, 4, 5, 5];
1841        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1842    }
1843
1844    fn one_shard_validation_fileinfo(block_size: usize) -> FileInfo {
1845        let mut fi = FileInfo::new("bucket/object", 1, 0);
1846        fi.erasure.index = 1;
1847        fi.erasure.block_size = block_size;
1848        fi
1849    }
1850
1851    #[test]
1852    fn validate_require_erasure_rejects_unrepresentable_shard_sizes() {
1853        let fi = one_shard_validation_fileinfo(usize::MAX);
1854        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1855
1856        if let Some((max_i64, above_i64_max)) = usize::try_from(i64::MAX)
1857            .ok()
1858            .and_then(|max_i64| max_i64.checked_add(1).map(|above_i64_max| (max_i64, above_i64_max)))
1859        {
1860            let mut fi = FileInfo::new("bucket/object", 2, 0);
1861            fi.erasure.index = 1;
1862            fi.erasure.block_size = above_i64_max;
1863            assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1864
1865            let mut fi = FileInfo::new("bucket/object", 2, 0);
1866            fi.erasure.index = 1;
1867            fi.erasure.block_size = max_i64;
1868            fi.parts = vec![ObjectPartInfo {
1869                number: 1,
1870                size: above_i64_max,
1871                ..Default::default()
1872            }];
1873            assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1874        }
1875
1876        let mut fi = one_shard_validation_fileinfo(2);
1877        fi.size = -1;
1878        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1879
1880        let mut fi = one_shard_validation_fileinfo(2);
1881        fi.size = i64::MAX;
1882        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1883
1884        let mut fi = one_shard_validation_fileinfo(2);
1885        fi.parts = vec![ObjectPartInfo {
1886            number: 1,
1887            size: usize::MAX,
1888            ..Default::default()
1889        }];
1890        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1891
1892        // A negative `actual_size` is the compressed "unknown size" sentinel, not an
1893        // unrepresentable size: it must stay readable so existing compressed objects
1894        // remain accessible. Only real (non-negative) sizes are shard-validated.
1895        let mut fi = one_shard_validation_fileinfo(2);
1896        fi.parts = vec![ObjectPartInfo {
1897            number: 1,
1898            actual_size: -1,
1899            ..Default::default()
1900        }];
1901        fi.validate(ValidationMode::RequireErasure)
1902            .expect("negative actual_size sentinel (compressed objects) must remain readable");
1903
1904        let mut fi = one_shard_validation_fileinfo(2);
1905        fi.parts = vec![ObjectPartInfo {
1906            number: 1,
1907            actual_size: i64::MAX,
1908            ..Default::default()
1909        }];
1910        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1911    }
1912
1913    #[test]
1914    fn metadata_read_validation_rejects_flags_that_try_to_relax_payload_validation() {
1915        let mut fi = FileInfo::new("bucket/legacy", 0, 2);
1916        fi.erasure.index = 1;
1917        fi.deleted = true;
1918        fi.transition_status = TRANSITION_COMPLETE.to_owned();
1919        fi.transitioned_objname = "remote/object".to_owned();
1920        fi.transition_tier = "WARM".to_owned();
1921
1922        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1923        assert!(!fi.is_valid(), "wire-controlled state flags must not relax is_valid");
1924        assert!(matches!(fi.validate_for_metadata_read(), Err(Error::FileCorrupt)));
1925        assert_file_corrupt(&fi, ValidationMode::DeleteOnly);
1926    }
1927
1928    #[test]
1929    fn metadata_read_validation_rejects_conflicting_transition_versions() {
1930        let mut fi = one_shard_validation_fileinfo(1);
1931        fi.transition_version_id = Some(Uuid::new_v4());
1932        fi.transition_version = Some(Uuid::new_v4().to_string());
1933
1934        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
1935    }
1936
1937    #[test]
1938    fn metadata_read_validation_requires_canonical_delete_marker_shape() {
1939        let marker = FileInfo {
1940            volume: "bucket".to_string(),
1941            name: "object".to_string(),
1942            version_id: Some(Uuid::new_v4()),
1943            deleted: true,
1944            mod_time: Some(OffsetDateTime::now_utc()),
1945            ..Default::default()
1946        };
1947        marker
1948            .validate_for_metadata_read()
1949            .expect("canonical delete marker should validate");
1950
1951        let mut missing_time = marker.clone();
1952        missing_time.mod_time = None;
1953        assert!(matches!(missing_time.validate_for_metadata_read(), Err(Error::FileCorrupt)));
1954
1955        let mut epoch_time = marker.clone();
1956        epoch_time.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
1957        assert!(matches!(epoch_time.validate_for_metadata_read(), Err(Error::FileCorrupt)));
1958
1959        let mut disguised_payload = marker.clone();
1960        disguised_payload.erasure.data_blocks = 1;
1961        assert!(matches!(disguised_payload.validate_for_metadata_read(), Err(Error::FileCorrupt)));
1962
1963        let mut marker_with_part = marker;
1964        marker_with_part.parts.push(ObjectPartInfo {
1965            number: 1,
1966            ..Default::default()
1967        });
1968        assert!(matches!(marker_with_part.validate_for_metadata_read(), Err(Error::FileCorrupt)));
1969
1970        let mut purge_pending = validation_test_fileinfo();
1971        purge_pending.deleted = true;
1972        purge_pending
1973            .validate_for_metadata_read()
1974            .expect("erasure-backed purge-pending objects remain valid payload metadata");
1975        assert!(!purge_pending.is_canonical_delete_marker());
1976    }
1977
1978    #[test]
1979    fn rename_validation_accepts_only_legacy_assigned_delete_marker_index() {
1980        let marker = FileInfo {
1981            volume: "bucket".to_string(),
1982            name: "object".to_string(),
1983            version_id: Some(Uuid::new_v4()),
1984            deleted: true,
1985            mod_time: Some(OffsetDateTime::now_utc()),
1986            ..Default::default()
1987        };
1988
1989        let mut legacy_marker = marker.clone();
1990        legacy_marker.erasure.index = 2;
1991        assert_eq!(legacy_marker.validate_for_metadata_read(), Err(Error::FileCorrupt));
1992        assert!(legacy_marker.is_legacy_indexed_delete_marker());
1993        legacy_marker.erasure.index = 0;
1994        legacy_marker
1995            .validate_for_metadata_read()
1996            .expect("normalized legacy marker should pass strict metadata validation");
1997
1998        let mut malformed_marker = legacy_marker;
1999        malformed_marker.erasure.index = 2;
2000        malformed_marker.erasure.data_blocks = 1;
2001        assert!(!malformed_marker.is_legacy_indexed_delete_marker());
2002        assert_eq!(malformed_marker.validate_for_metadata_read(), Err(Error::FileCorrupt));
2003
2004        let mut forged_free_marker = marker.clone();
2005        insert_str(&mut forged_free_marker.metadata, SUFFIX_FREE_VERSION, String::new());
2006        assert_eq!(forged_free_marker.validate_for_metadata_read(), Err(Error::FileCorrupt));
2007
2008        let mut free_marker = marker.clone();
2009        insert_str(&mut free_marker.metadata, SUFFIX_FREE_VERSION, String::new());
2010        free_marker.set_tier_free_version();
2011        free_marker.transition_tier = "WARM".to_string();
2012        free_marker.transitioned_objname = "remote/object".to_string();
2013        free_marker
2014            .validate_for_metadata_read()
2015            .expect("complete tier free-version marker should remain valid");
2016
2017        let mut out_of_range_marker = marker;
2018        out_of_range_marker.erasure.index = MAX_ERASURE_SHARDS + 1;
2019        assert!(!out_of_range_marker.is_legacy_indexed_delete_marker());
2020        assert_eq!(out_of_range_marker.validate_for_metadata_read(), Err(Error::FileCorrupt));
2021    }
2022
2023    #[test]
2024    fn delete_operation_accepts_only_the_worker_tier_free_version_request_shape() {
2025        let mut worker_request = FileInfo {
2026            volume: "bucket".to_string(),
2027            name: "object".to_string(),
2028            version_id: Some(Uuid::new_v4()),
2029            deleted: true,
2030            ..Default::default()
2031        };
2032        worker_request.set_tier_free_version();
2033        worker_request
2034            .validate_for_delete_operation()
2035            .expect("the lifecycle worker request shape must remain valid");
2036
2037        let mut timestamped = worker_request.clone();
2038        timestamped.mod_time = Some(OffsetDateTime::now_utc());
2039        assert_eq!(timestamped.validate_for_delete_operation(), Err(Error::FileCorrupt));
2040
2041        let mut with_payload = worker_request.clone();
2042        with_payload.size = 1;
2043        assert_eq!(with_payload.validate_for_delete_operation(), Err(Error::FileCorrupt));
2044
2045        let mut with_unrelated_metadata = worker_request;
2046        with_unrelated_metadata
2047            .metadata
2048            .insert("unexpected".to_string(), "value".to_string());
2049        assert_eq!(with_unrelated_metadata.validate_for_delete_operation(), Err(Error::FileCorrupt));
2050    }
2051
2052    #[test]
2053    fn erasure_and_delete_operation_validation_enforce_basic_collection_bounds() {
2054        let mut fi = validation_test_fileinfo();
2055        fi.parts = (1..=10_001)
2056            .map(|number| ObjectPartInfo {
2057                number,
2058                ..Default::default()
2059            })
2060            .collect();
2061        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2062        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2063
2064        let mut fi = validation_test_fileinfo();
2065        fi.erasure.checksums = (1..=10_001)
2066            .map(|part_number| ChecksumInfo {
2067                part_number,
2068                ..Default::default()
2069            })
2070            .collect();
2071        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2072        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2073
2074        let mut fi = validation_test_fileinfo();
2075        fi.erasure.distribution = vec![1; 17];
2076        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2077        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2078    }
2079
2080    #[test]
2081    fn validate_accepts_exact_collection_limits() {
2082        let mut fi = validation_test_fileinfo();
2083        fi.parts = (1..=10_000)
2084            .map(|number| ObjectPartInfo {
2085                number,
2086                ..Default::default()
2087            })
2088            .collect();
2089        fi.erasure.checksums = (1..=10_000)
2090            .map(|part_number| ChecksumInfo {
2091                part_number,
2092                ..Default::default()
2093            })
2094            .collect();
2095
2096        fi.validate(ValidationMode::RequireErasure)
2097            .expect("exact collection limits must remain valid");
2098        fi.validate_for_delete_operation()
2099            .expect("non-marker delete requests must accept exact collection limits");
2100    }
2101
2102    #[test]
2103    fn erasure_and_delete_operation_validation_reject_checksum_association_errors() {
2104        let mut fi = validation_test_fileinfo();
2105        fi.erasure.checksums[0].part_number = 99;
2106        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2107        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2108
2109        let mut fi = validation_test_fileinfo();
2110        let duplicate = fi.erasure.checksums[0].clone();
2111        fi.erasure.checksums.push(duplicate);
2112        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2113        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2114    }
2115
2116    #[test]
2117    fn erasure_and_delete_operation_validation_reject_duplicate_parts() {
2118        let mut fi = validation_test_fileinfo();
2119        fi.parts[1].number = fi.parts[0].number;
2120        fi.erasure.checksums.clear();
2121
2122        assert_file_corrupt(&fi, ValidationMode::RequireErasure);
2123        assert_eq!(fi.validate_for_delete_operation(), Err(Error::FileCorrupt));
2124    }
2125
2126    #[test]
2127    fn write_quorum_distinguishes_purge_pending_payload_from_delete_marker() {
2128        let mut purge_pending = FileInfo::new("bucket/object", 5, 1);
2129        purge_pending.erasure.index = 1;
2130        purge_pending.deleted = true;
2131        assert_eq!(purge_pending.write_quorum(4), 5);
2132
2133        let marker = FileInfo {
2134            deleted: true,
2135            mod_time: Some(OffsetDateTime::now_utc()),
2136            ..Default::default()
2137        };
2138        assert_eq!(marker.write_quorum(4), 4);
2139
2140        let malformed_marker = FileInfo {
2141            deleted: true,
2142            ..Default::default()
2143        };
2144        assert_eq!(malformed_marker.write_quorum(4), 4, "invalid marker metadata must never reduce quorum");
2145    }
2146
2147    fn small_string_strategy() -> impl Strategy<Value = String> {
2148        proptest::string::string_regex("[A-Za-z0-9._/-]{0,16}").expect("small string regex should compile")
2149    }
2150
2151    fn small_required_string_strategy() -> impl Strategy<Value = String> {
2152        proptest::string::string_regex("[A-Za-z0-9._/-]{1,16}").expect("required string regex should compile")
2153    }
2154
2155    fn bytes_strategy(max_len: usize) -> impl Strategy<Value = Bytes> {
2156        vec(any::<u8>(), 0..=max_len).prop_map(Bytes::from)
2157    }
2158
2159    fn uuid_strategy() -> impl Strategy<Value = Uuid> {
2160        any::<[u8; 16]>().prop_map(Uuid::from_bytes)
2161    }
2162
2163    fn optional_uuid_strategy() -> impl Strategy<Value = Option<Uuid>> {
2164        proptest::option::of(uuid_strategy())
2165    }
2166
2167    fn timestamp_strategy() -> impl Strategy<Value = OffsetDateTime> {
2168        (0i64..=4_102_444_800i64)
2169            .prop_map(|seconds| OffsetDateTime::from_unix_timestamp(seconds).expect("bounded timestamp should be valid"))
2170    }
2171
2172    fn optional_timestamp_strategy() -> impl Strategy<Value = Option<OffsetDateTime>> {
2173        proptest::option::of(timestamp_strategy())
2174    }
2175
2176    fn hash_algorithm_strategy() -> impl Strategy<Value = HashAlgorithm> {
2177        prop_oneof![
2178            Just(HashAlgorithm::SHA256),
2179            Just(HashAlgorithm::HighwayHash256),
2180            Just(HashAlgorithm::HighwayHash256S),
2181            Just(HashAlgorithm::HighwayHash256SLegacy),
2182            Just(HashAlgorithm::BLAKE2b512),
2183            Just(HashAlgorithm::Md5),
2184            Just(HashAlgorithm::None),
2185        ]
2186    }
2187
2188    fn checksum_info_strategy() -> impl Strategy<Value = ChecksumInfo> {
2189        (0usize..=4, hash_algorithm_strategy(), bytes_strategy(32)).prop_map(|(part_number, algorithm, hash)| ChecksumInfo {
2190            part_number,
2191            algorithm,
2192            hash,
2193        })
2194    }
2195
2196    fn erasure_info_strategy() -> impl Strategy<Value = ErasureInfo> {
2197        (1usize..=4, 0usize..=3, 2usize..=4096)
2198            .prop_flat_map(|(data_blocks, parity_blocks, block_size)| {
2199                let total_blocks = data_blocks + parity_blocks;
2200                (
2201                    Just(data_blocks),
2202                    Just(parity_blocks),
2203                    Just(block_size),
2204                    0usize..=total_blocks,
2205                    vec(1usize..=(total_blocks.max(1)), total_blocks),
2206                    vec(checksum_info_strategy(), 0..=3),
2207                    small_required_string_strategy(),
2208                )
2209            })
2210            .prop_map(
2211                |(data_blocks, parity_blocks, block_size, index, distribution, checksums, algorithm)| ErasureInfo {
2212                    algorithm,
2213                    data_blocks,
2214                    parity_blocks,
2215                    block_size,
2216                    index,
2217                    distribution,
2218                    checksums,
2219                },
2220            )
2221    }
2222
2223    fn object_part_info_strategy() -> impl Strategy<Value = ObjectPartInfo> {
2224        (
2225            small_string_strategy(),
2226            0usize..=8,
2227            0usize..=4096,
2228            -1_000_000i64..=1_000_000i64,
2229            optional_timestamp_strategy(),
2230            proptest::option::of(bytes_strategy(16)),
2231            proptest::option::of(hash_map(small_string_strategy(), small_string_strategy(), 0..=3)),
2232            proptest::option::of(small_string_strategy()),
2233        )
2234            .prop_map(|(etag, number, size, actual_size, mod_time, index, checksums, error)| ObjectPartInfo {
2235                etag,
2236                number,
2237                size,
2238                actual_size,
2239                mod_time,
2240                index,
2241                checksums,
2242                error,
2243                integrity: None,
2244            })
2245    }
2246
2247    fn file_info_strategy() -> impl Strategy<Value = FileInfo> {
2248        (
2249            (
2250                small_string_strategy(),
2251                small_string_strategy(),
2252                optional_uuid_strategy(),
2253                any::<bool>(),
2254                any::<bool>(),
2255                small_string_strategy(),
2256                small_string_strategy(),
2257                small_string_strategy(),
2258                optional_uuid_strategy(),
2259                any::<bool>(),
2260            ),
2261            (
2262                optional_uuid_strategy(),
2263                optional_timestamp_strategy(),
2264                -1_000_000i64..=1_000_000i64,
2265                proptest::option::of(any::<u32>()),
2266                proptest::option::of(any::<u64>()),
2267                hash_map(small_string_strategy(), small_string_strategy(), 0..=4),
2268                vec(object_part_info_strategy(), 0..=3),
2269                erasure_info_strategy(),
2270                any::<bool>(),
2271                proptest::option::of(bytes_strategy(32)),
2272            ),
2273            (
2274                0usize..=4,
2275                optional_timestamp_strategy(),
2276                any::<bool>(),
2277                0usize..=4,
2278                proptest::option::of(bytes_strategy(32)),
2279                any::<bool>(),
2280                any::<bool>(),
2281            ),
2282        )
2283            .prop_map(|(head, middle, tail)| {
2284                let (
2285                    volume,
2286                    name,
2287                    version_id,
2288                    is_latest,
2289                    deleted,
2290                    transition_status,
2291                    transitioned_objname,
2292                    transition_tier,
2293                    transition_version_id,
2294                    expire_restored,
2295                ) = head;
2296                let (data_dir, mod_time, size, mode, written_by_version, metadata, parts, erasure, mark_deleted, data) = middle;
2297                let (num_versions, successor_mod_time, fresh, idx, checksum, versioned, uses_legacy_checksum) = tail;
2298
2299                FileInfo {
2300                    volume,
2301                    name,
2302                    version_id,
2303                    is_latest,
2304                    deleted,
2305                    transition_status,
2306                    transitioned_objname,
2307                    transition_tier,
2308                    transition_version_id,
2309                    transition_version: transition_version_id.map(|version_id| version_id.to_string()),
2310                    transition_version_state: if transition_version_id.is_some() {
2311                        TransitionVersionState::Exact
2312                    } else {
2313                        TransitionVersionState::Unknown
2314                    },
2315                    expire_restored,
2316                    data_dir,
2317                    mod_time,
2318                    size,
2319                    mode,
2320                    written_by_version,
2321                    metadata,
2322                    parts,
2323                    erasure,
2324                    mark_deleted,
2325                    replication_state_internal: None,
2326                    data,
2327                    num_versions,
2328                    successor_mod_time,
2329                    fresh,
2330                    idx,
2331                    checksum,
2332                    versioned,
2333                    uses_legacy_checksum,
2334                    overwrite_tier_free_version_id: None,
2335                }
2336            })
2337    }
2338
2339    fn positional_compat_file_info() -> FileInfo {
2340        FileInfo {
2341            volume: "bucket".to_string(),
2342            name: "object".to_string(),
2343            transition_version_id: Some(Uuid::from_u128(1)),
2344            transition_version: Some(Uuid::from_u128(1).to_string()),
2345            transition_version_state: TransitionVersionState::Exact,
2346            expire_restored: true,
2347            size: -1,
2348            fresh: true,
2349            idx: 7,
2350            versioned: true,
2351            uses_legacy_checksum: true,
2352            ..Default::default()
2353        }
2354    }
2355
2356    #[derive(Clone, Copy)]
2357    enum HistoricalFileInfoLayout {
2358        Beta11,
2359        Beta12,
2360    }
2361
2362    fn encode_historical_file_info(value: &FileInfo, layout: HistoricalFileInfoLayout) -> Vec<u8> {
2363        let mut encoded = Vec::new();
2364        let field_count = match layout {
2365            HistoricalFileInfoLayout::Beta11 => 28,
2366            HistoricalFileInfoLayout::Beta12 => 30,
2367        };
2368        rmp::encode::write_array_len(&mut encoded, field_count).expect("historical FileInfo array header should encode");
2369
2370        macro_rules! encode_fields {
2371            ($($field:expr),+ $(,)?) => {
2372                $($field
2373                    .serialize(&mut Serializer::new(&mut encoded))
2374                    .expect("historical FileInfo field should encode");)+
2375            };
2376        }
2377
2378        encode_fields!(
2379            &value.volume,
2380            &value.name,
2381            &value.version_id,
2382            &value.is_latest,
2383            &value.deleted,
2384            &value.transition_status,
2385            &value.transitioned_objname,
2386            &value.transition_tier,
2387            &value.transition_version_id,
2388        );
2389        if matches!(layout, HistoricalFileInfoLayout::Beta12) {
2390            encode_fields!(&value.transition_version, &value.transition_version_state);
2391        }
2392        encode_fields!(
2393            &value.expire_restored,
2394            &value.data_dir,
2395            &value.mod_time,
2396            &value.size,
2397            &value.mode,
2398            &value.written_by_version,
2399            &value.metadata,
2400            &value.parts,
2401            &value.erasure,
2402            &value.mark_deleted,
2403            &value.replication_state_internal,
2404            &value.data,
2405            &value.num_versions,
2406            &value.successor_mod_time,
2407            &value.fresh,
2408            &value.idx,
2409            &value.checksum,
2410            &value.versioned,
2411            &value.uses_legacy_checksum,
2412        );
2413        encoded
2414    }
2415
2416    fn field_order_file_info() -> FileInfo {
2417        let transition_version_id = Uuid::from_u128(12);
2418        FileInfo {
2419            volume: "volume-sentinel".to_string(),
2420            name: "name-sentinel".to_string(),
2421            version_id: Some(Uuid::from_u128(11)),
2422            is_latest: true,
2423            deleted: false,
2424            transition_status: "transition-status".to_string(),
2425            transitioned_objname: "transitioned-object".to_string(),
2426            transition_tier: "transition-tier".to_string(),
2427            transition_version_id: Some(transition_version_id),
2428            transition_version: Some(transition_version_id.to_string()),
2429            transition_version_state: TransitionVersionState::Exact,
2430            expire_restored: true,
2431            data_dir: Some(Uuid::from_u128(13)),
2432            mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(14)),
2433            size: 15,
2434            mode: Some(16),
2435            written_by_version: Some(17),
2436            metadata: [("metadata-key".to_string(), "metadata-value".to_string())]
2437                .into_iter()
2438                .collect(),
2439            parts: vec![ObjectPartInfo {
2440                etag: "part-etag".to_string(),
2441                number: 18,
2442                size: 19,
2443                actual_size: 20,
2444                mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(21)),
2445                index: Some(Bytes::from_static(b"part-index")),
2446                checksums: Some(
2447                    [("part-checksum".to_string(), "checksum-value".to_string())]
2448                        .into_iter()
2449                        .collect(),
2450                ),
2451                error: Some("part-error".to_string()),
2452                integrity: None,
2453            }],
2454            erasure: ErasureInfo {
2455                algorithm: "erasure-algorithm".to_string(),
2456                data_blocks: 2,
2457                parity_blocks: 1,
2458                block_size: 1024,
2459                index: 3,
2460                distribution: vec![2, 1, 3],
2461                checksums: vec![ChecksumInfo {
2462                    part_number: 22,
2463                    algorithm: HashAlgorithm::SHA256,
2464                    hash: Bytes::from_static(b"erasure-hash"),
2465                }],
2466            },
2467            mark_deleted: false,
2468            replication_state_internal: Some(ReplicationState {
2469                replicate_decision_str: "replication-decision".to_string(),
2470                delete_marker: true,
2471                ..Default::default()
2472            }),
2473            data: Some(Bytes::from_static(b"inline-data")),
2474            num_versions: 23,
2475            successor_mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(24)),
2476            fresh: true,
2477            idx: 25,
2478            checksum: Some(Bytes::from_static(b"combined-checksum")),
2479            versioned: false,
2480            uses_legacy_checksum: true,
2481            overwrite_tier_free_version_id: None,
2482        }
2483    }
2484
2485    const BETA11_FILEINFO_FIXTURE: &[u8] = &[
2486        220, 0, 28, 166, 98, 117, 99, 107, 101, 116, 166, 111, 98, 106, 101, 99, 116, 192, 194, 194, 160, 160, 160, 196, 16, 0,
2487        0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 195, 192, 192, 255, 192, 192, 128, 144, 151, 160, 0, 0, 0, 0, 144, 144, 194,
2488        192, 192, 0, 192, 195, 7, 192, 195, 195,
2489    ];
2490
2491    const BETA12_FILEINFO_FIXTURE: &[u8] = &[
2492        220, 0, 30, 166, 98, 117, 99, 107, 101, 116, 166, 111, 98, 106, 101, 99, 116, 192, 194, 194, 160, 160, 160, 196, 16, 0,
2493        0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 217, 36, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48,
2494        48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48, 48, 48, 48, 48, 48, 48, 49, 165, 101, 120, 97, 99, 116, 195, 192, 192,
2495        255, 192, 192, 128, 144, 151, 160, 0, 0, 0, 0, 144, 144, 194, 192, 192, 0, 192, 195, 7, 192, 195, 195,
2496    ];
2497
2498    #[test]
2499    fn fileinfo_decodes_beta11_positional_layout() {
2500        let expected = positional_compat_file_info();
2501        let decoded = FileInfo::unmarshal(BETA11_FILEINFO_FIXTURE).expect("beta.11 positional FileInfo should decode");
2502
2503        assert_eq!(decoded.volume, expected.volume);
2504        assert_eq!(decoded.expire_restored, expected.expire_restored);
2505        assert_eq!(decoded.idx, expected.idx);
2506        assert_eq!(decoded.transition_version, None);
2507        assert_eq!(decoded.transition_version_state, TransitionVersionState::Unknown);
2508    }
2509
2510    #[test]
2511    fn fileinfo_decodes_beta12_positional_layout() {
2512        let expected = positional_compat_file_info();
2513        let decoded = FileInfo::unmarshal(BETA12_FILEINFO_FIXTURE).expect("beta.12 positional FileInfo should decode");
2514
2515        assert_eq!(decoded, expected);
2516    }
2517
2518    #[test]
2519    fn fileinfo_historical_layouts_preserve_all_field_positions() {
2520        let expected = field_order_file_info();
2521        let beta12 = encode_historical_file_info(&expected, HistoricalFileInfoLayout::Beta12);
2522        assert_eq!(FileInfo::unmarshal(&beta12).expect("beta.12 full-field FileInfo should decode"), expected);
2523
2524        let beta11 = encode_historical_file_info(&expected, HistoricalFileInfoLayout::Beta11);
2525        let mut beta11_expected = expected;
2526        beta11_expected.transition_version = None;
2527        beta11_expected.transition_version_state = TransitionVersionState::Unknown;
2528        assert_eq!(
2529            FileInfo::unmarshal(&beta11).expect("beta.11 full-field FileInfo should decode"),
2530            beta11_expected
2531        );
2532    }
2533
2534    #[test]
2535    fn fileinfo_rejects_unsupported_positional_lengths() {
2536        let mut fields_29 = BETA11_FILEINFO_FIXTURE.to_vec();
2537        fields_29[2] = 29;
2538        fields_29.push(0xc0);
2539        let error = FileInfo::unmarshal(&fields_29).expect_err("29-field FileInfo must fail closed");
2540        assert!(matches!(error, Error::RmpSerdeDecode(message) if message.contains("invalid length 29")));
2541
2542        let mut fields_31 = BETA12_FILEINFO_FIXTURE.to_vec();
2543        fields_31[2] = 31;
2544        fields_31.push(0xc0);
2545        let error = FileInfo::unmarshal(&fields_31).expect_err("31-field FileInfo must fail closed");
2546        assert!(matches!(error, Error::RmpSerdeDecode(message) if message.contains("invalid length 31")));
2547    }
2548
2549    #[test]
2550    fn fileinfo_rejects_truncated_historical_layouts() {
2551        let beta11 = &BETA11_FILEINFO_FIXTURE[..BETA11_FILEINFO_FIXTURE.len() - 1];
2552        assert!(matches!(FileInfo::unmarshal(beta11), Err(Error::RmpSerdeDecode(_))));
2553
2554        let state_offset = BETA12_FILEINFO_FIXTURE
2555            .windows(6)
2556            .position(|window| window == [0xa5, b'e', b'x', b'a', b'c', b't'])
2557            .expect("beta.12 fixture should contain the exact transition state");
2558        assert!(matches!(
2559            FileInfo::unmarshal(&BETA12_FILEINFO_FIXTURE[..state_offset]),
2560            Err(Error::RmpSerdeDecode(_))
2561        ));
2562    }
2563
2564    fn wrap_historical_file_info(fixture: &[u8]) -> Vec<u8> {
2565        let mut encoded = Vec::new();
2566        rmp::encode::write_array_len(&mut encoded, 5).expect("FileInfoVersions array header should encode");
2567        "bucket"
2568            .serialize(&mut Serializer::new(&mut encoded))
2569            .expect("FileInfoVersions volume should encode");
2570        "object"
2571            .serialize(&mut Serializer::new(&mut encoded))
2572            .expect("FileInfoVersions name should encode");
2573        Option::<OffsetDateTime>::None
2574            .serialize(&mut Serializer::new(&mut encoded))
2575            .expect("FileInfoVersions mod time should encode");
2576        rmp::encode::write_array_len(&mut encoded, 1).expect("FileInfoVersions versions header should encode");
2577        encoded.extend_from_slice(fixture);
2578        rmp::encode::write_array_len(&mut encoded, 0).expect("FileInfoVersions free versions header should encode");
2579        encoded
2580    }
2581
2582    #[test]
2583    fn fileinfo_versions_decodes_nested_historical_layouts() {
2584        for fixture in [BETA11_FILEINFO_FIXTURE, BETA12_FILEINFO_FIXTURE] {
2585            let encoded = wrap_historical_file_info(fixture);
2586            let decoded: FileInfoVersions =
2587                rmp_serde::from_slice(&encoded).expect("nested historical FileInfo should decode through FileInfoVersions");
2588            assert_eq!(decoded.versions.len(), 1);
2589            assert_eq!(decoded.versions[0].name, "object");
2590            assert!(decoded.free_versions.is_empty());
2591        }
2592    }
2593
2594    #[derive(Deserialize)]
2595    struct Beta11MapProbe {
2596        volume: String,
2597        expire_restored: bool,
2598        idx: usize,
2599    }
2600
2601    #[derive(Deserialize)]
2602    struct Beta12MapProbe {
2603        transition_version: Option<String>,
2604        transition_version_state: TransitionVersionState,
2605        expire_restored: bool,
2606    }
2607
2608    #[derive(Deserialize)]
2609    struct Beta11NestedMapProbe {
2610        file_info: Beta11MapProbe,
2611    }
2612
2613    #[derive(Serialize)]
2614    struct NestedFileInfo<'a> {
2615        file_info: &'a FileInfo,
2616    }
2617
2618    #[test]
2619    fn fileinfo_serializes_as_map_readable_by_beta11_and_beta12_shapes() {
2620        let mut expected = positional_compat_file_info();
2621        expected.overwrite_tier_free_version_id = Some(Uuid::new_v4());
2622        let encoded = expected.marshal_msg().expect("current FileInfo map should encode");
2623        let mut cursor = encoded.as_slice();
2624        let field_count = usize::try_from(rmp::decode::read_map_len(&mut cursor).expect("FileInfo should start with a map"))
2625            .expect("FileInfo map field count should fit usize");
2626        assert_eq!(field_count, FILE_INFO_FIELDS.len());
2627        let current = FileInfo::unmarshal(&encoded).expect("current reader preserves the transient PUT intent");
2628        assert_eq!(current.overwrite_tier_free_version_id, expected.overwrite_tier_free_version_id);
2629
2630        let beta11: Beta11MapProbe = rmp_serde::from_slice(&encoded).expect("beta.11 field shape should read current map");
2631        assert_eq!(beta11.volume, expected.volume);
2632        assert_eq!(beta11.expire_restored, expected.expire_restored);
2633        assert_eq!(beta11.idx, expected.idx);
2634
2635        let beta12: Beta12MapProbe = rmp_serde::from_slice(&encoded).expect("beta.12 field shape should read current map");
2636        assert_eq!(beta12.transition_version, expected.transition_version);
2637        assert_eq!(beta12.transition_version_state, expected.transition_version_state);
2638        assert_eq!(beta12.expire_restored, expected.expire_restored);
2639
2640        let nested = NestedFileInfo { file_info: &expected };
2641        let nested_encoded = rmp_serde::to_vec(&nested).expect("nested FileInfo should encode");
2642        let nested_beta11: Beta11NestedMapProbe =
2643            rmp_serde::from_slice(&nested_encoded).expect("beta.11 field shape should read nested current map");
2644        assert_eq!(nested_beta11.file_info.volume, expected.volume);
2645        assert_eq!(nested_beta11.file_info.expire_restored, expected.expire_restored);
2646    }
2647
2648    #[test]
2649    fn fileinfo_defaults_missing_put_intent_without_extending_legacy_arrays() {
2650        let expected = positional_compat_file_info();
2651        let encoded = expected.marshal_msg().expect("ordinary FileInfo map should encode");
2652        let mut cursor = encoded.as_slice();
2653        assert_eq!(
2654            rmp::decode::read_map_len(&mut cursor).expect("FileInfo map header") as usize,
2655            FILE_INFO_FIELDS.len() - 1
2656        );
2657        assert!(
2658            FileInfo::unmarshal(&encoded)
2659                .expect("map without PUT intent")
2660                .overwrite_tier_free_version_id
2661                .is_none()
2662        );
2663        for fixture in [BETA11_FILEINFO_FIXTURE, BETA12_FILEINFO_FIXTURE] {
2664            assert!(
2665                FileInfo::unmarshal(fixture)
2666                    .expect("historical positional array")
2667                    .overwrite_tier_free_version_id
2668                    .is_none()
2669            );
2670        }
2671    }
2672
2673    #[test]
2674    fn fileinfo_deserializes_beta11_map_without_transition_fields() {
2675        let expected = positional_compat_file_info();
2676        let mut value = serde_json::to_value(&expected).expect("current FileInfo should serialize to JSON");
2677        let object = value.as_object_mut().expect("FileInfo JSON should be an object");
2678        object.remove("transition_version");
2679        object.remove("transition_version_state");
2680
2681        let decoded: FileInfo = serde_json::from_value(value).expect("beta.11 FileInfo JSON should decode");
2682        assert_eq!(decoded.volume, expected.volume);
2683        assert_eq!(decoded.expire_restored, expected.expire_restored);
2684        assert_eq!(decoded.transition_version, None);
2685        assert_eq!(decoded.transition_version_state, TransitionVersionState::Unknown);
2686    }
2687
2688    proptest! {
2689        #[test]
2690        fn fileinfo_msgpack_round_trips(value in file_info_strategy()) {
2691            let encoded = value
2692                .marshal_msg()
2693                .expect("FileInfo should serialize in the property roundtrip test");
2694            let decoded = FileInfo::unmarshal(&encoded)
2695                .expect("FileInfo should deserialize in the property roundtrip test");
2696
2697            prop_assert_eq!(decoded, value);
2698        }
2699
2700        #[test]
2701        fn fileinfo_unmarshal_never_panics(input in vec(any::<u8>(), 0..=1024)) {
2702            let result = std::panic::catch_unwind(|| FileInfo::unmarshal(&input));
2703            prop_assert!(result.is_ok(), "FileInfo::unmarshal panicked for arbitrary input");
2704        }
2705    }
2706
2707    #[test]
2708    fn replication_info_equals_compares_mark_deleted_and_replication_state() {
2709        let base = FileInfo::default();
2710
2711        // Identical (both default) infos are equal.
2712        assert!(base.replication_info_equals(&FileInfo::default()));
2713
2714        // Differing mark_deleted breaks equality.
2715        let marked = FileInfo {
2716            mark_deleted: true,
2717            ..Default::default()
2718        };
2719        assert!(!base.replication_info_equals(&marked));
2720
2721        // Differing replication_state_internal breaks equality (regression guard:
2722        // this field used to be ignored by replication_info_equals).
2723        let with_state = FileInfo {
2724            replication_state_internal: Some(ReplicationState {
2725                replicate_decision_str: "arn:aws:s3:::dest".to_string(),
2726                ..Default::default()
2727            }),
2728            ..Default::default()
2729        };
2730        assert!(!base.replication_info_equals(&with_state));
2731
2732        // Equal replication states remain equal.
2733        let with_state_clone = FileInfo {
2734            replication_state_internal: with_state.replication_state_internal.clone(),
2735            ..Default::default()
2736        };
2737        assert!(with_state.replication_info_equals(&with_state_clone));
2738    }
2739
2740    #[test]
2741    fn debug_redacts_sealed_encryption_metadata_values() {
2742        let sealed_key = "IAAfANqt7wIJfVSgFAG3f5S6HuC2eyM5DdJlx7RSJKw2ZakSb3d5";
2743        let sealed_iv = "0Vr8QLGvQThk8gIWFCUnBOTUwZgs7TTBteRnAK9avD0=";
2744        let mut fi = FileInfo::default();
2745        for key in [
2746            "X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key",
2747            "X-Minio-Internal-Server-Side-Encryption-Sealed-Key",
2748        ] {
2749            fi.metadata.insert(key.to_string(), sealed_key.to_string());
2750        }
2751        for key in [
2752            "X-Rustfs-Internal-Server-Side-Encryption-Iv",
2753            "X-Minio-Internal-Server-Side-Encryption-Iv",
2754            "x-rustfs-encryption-iv",
2755        ] {
2756            fi.metadata.insert(key.to_string(), sealed_iv.to_string());
2757        }
2758        fi.metadata.insert("content-type".to_string(), "text/plain".to_string());
2759
2760        let dump = format!("{fi:?}");
2761        assert!(!dump.contains(sealed_key), "sealed key leaked into Debug output: {dump}");
2762        assert!(!dump.contains(sealed_iv), "sealed IV leaked into Debug output: {dump}");
2763        // Keys stay visible so operators can still see which metadata is present.
2764        assert!(dump.contains("X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key"));
2765        assert!(dump.contains(&format!("<redacted {} bytes>", sealed_key.len())));
2766        // Non-sensitive metadata values keep their diagnostic value.
2767        assert!(dump.contains("text/plain"));
2768    }
2769
2770    #[test]
2771    fn debug_redacts_replication_sse_transport_metadata_values() {
2772        let sealed_key = "IAAfANqt7wIJfVSgFAG3f5S6HuC2eyM5DdJlx7RSJKw2ZakSb3d5";
2773        let mut fi = FileInfo::default();
2774        for key in [
2775            "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key",
2776            "X-Rustfs-Replication-Server-Side-Encryption-Iv",
2777            "X-Rustfs-Replication-Encryption-Iv",
2778            "X-Rustfs-Replication-Ssec-Key-Md5",
2779        ] {
2780            fi.metadata.insert(key.to_string(), sealed_key.to_string());
2781        }
2782        fi.metadata.insert("content-type".to_string(), "text/plain".to_string());
2783
2784        let dump = format!("{fi:?}");
2785        assert!(
2786            !dump.contains(sealed_key),
2787            "replication SSE transport value leaked into Debug output: {dump}"
2788        );
2789        assert!(dump.contains("X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key"));
2790        assert!(dump.contains(&format!("<redacted {} bytes>", sealed_key.len())));
2791        assert!(dump.contains("text/plain"));
2792    }
2793
2794    #[test]
2795    fn debug_elides_inline_data_bytes() {
2796        let fi = FileInfo {
2797            data: Some(Bytes::from_static(b"plaintext user object content")),
2798            checksum: Some(Bytes::from_static(b"\x01\x02checksumblob")),
2799            ..Default::default()
2800        };
2801
2802        let dump = format!("{fi:?}");
2803        assert!(
2804            !dump.contains("plaintext user object content"),
2805            "inline data leaked into Debug output: {dump}"
2806        );
2807        assert!(!dump.contains("checksumblob"), "checksum bytes leaked into Debug output: {dump}");
2808        assert!(dump.contains("data: Some(<29 bytes elided>)"), "missing data length summary: {dump}");
2809
2810        let empty = FileInfo::default();
2811        assert!(format!("{empty:?}").contains("data: None"));
2812    }
2813
2814    /// backlog#1342: MinIO writes the persisted `x-amz-restore` completion
2815    /// header with Go's `http.TimeFormat` (RFC1123/GMT). Before this parser
2816    /// accepted that shape, every object migrated in place from MinIO failed
2817    /// `parse_restore_obj_status`.
2818    #[test]
2819    fn parses_minio_rfc1123_restore_expiry() {
2820        let status = parse_restore_obj_status("ongoing-request=\"false\", expiry-date=\"Wed, 01 Jan 2025 10:20:30 GMT\"")
2821            .expect("RFC1123 expiry-date must parse");
2822        assert_eq!(status.is_restore_in_progress, Some(false));
2823        let expiry = status.expiry().expect("completed restore carries an expiry");
2824        assert_eq!(expiry, datetime!(2025-01-01 10:20:30 UTC));
2825    }
2826
2827    /// The RustFS-written shape must keep parsing unchanged: the persisted
2828    /// format stays RFC3339 until every node can read both.
2829    #[test]
2830    fn parses_rustfs_rfc3339_restore_expiry() {
2831        let status = parse_restore_obj_status("ongoing-request=\"false\", expiry-date=\"2025-01-01T10:20:30Z\"")
2832            .expect("RFC3339 expiry-date must parse");
2833        assert_eq!(status.expiry(), Some(datetime!(2025-01-01 10:20:30 UTC)));
2834    }
2835
2836    /// Round-trip both directions of the migration: what RustFS emits on the
2837    /// wire (`to_string2`, RFC1123) and what it persists (`to_string`,
2838    /// RFC3339) must both be readable by this parser.
2839    #[test]
2840    fn restore_status_round_trips_through_both_formats() {
2841        let status = RestoreStatus {
2842            is_restore_in_progress: Some(false),
2843            restore_expiry_date: Some(datetime!(2030-06-15 07:08:09 UTC)),
2844        };
2845        for rendered in [RestoreStatusOps::to_string(&status), status.to_string2()] {
2846            let parsed = parse_restore_obj_status(&rendered).unwrap_or_else(|e| panic!("{rendered} must parse: {e}"));
2847            assert_eq!(parsed.expiry(), Some(datetime!(2030-06-15 07:08:09 UTC)), "{rendered}");
2848        }
2849    }
2850
2851    /// backlog#1735 A3c: filemeta's own `RestoreStatus` must persist the same
2852    /// bytes the s3s DTO did. The restore value in the pre-`metadata_keys`
2853    /// fixture was rendered by the s3s-backed writer; parsing and rendering it
2854    /// again must reproduce it exactly.
2855    #[test]
2856    fn restore_status_rerenders_pre_module_fixture_bytes() {
2857        let fi = crate::FileMeta::load(&crate::test_data::create_pre_metadata_keys_xlmeta().expect("decode fixture hex"))
2858            .expect("load fixture xl.meta")
2859            .into_fileinfo("bucket", "object", "0b1e5a3a-1735-4a3a-8000-00000000a3a0", false, false, false)
2860            .expect("fixture version to FileInfo");
2861        let stored = fi.metadata.get(metadata_keys::RESTORE).expect("fixture restore value");
2862        assert_eq!(stored, "ongoing-request=\"false\", expiry-date=\"9999-01-01T00:00:00Z\"");
2863
2864        let parsed = parse_restore_obj_status(stored).expect("fixture restore value parses");
2865        assert_eq!(
2866            parsed,
2867            RestoreStatus {
2868                is_restore_in_progress: Some(false),
2869                restore_expiry_date: Some(datetime!(9999-01-01 00:00:00 UTC)),
2870            }
2871        );
2872        assert_eq!(RestoreStatusOps::to_string(&parsed), *stored);
2873        assert_eq!(
2874            parsed.to_string2(),
2875            "ongoing-request=\"false\", expiry-date=\"Fri, 01 Jan 9999 00:00:00 GMT\""
2876        );
2877
2878        let ongoing = RestoreStatus {
2879            is_restore_in_progress: Some(true),
2880            restore_expiry_date: Some(datetime!(2030-06-15 07:08:09 UTC)),
2881        };
2882        assert_eq!(RestoreStatusOps::to_string(&ongoing), "ongoing-request=\"true\"");
2883        assert_eq!(
2884            parse_restore_obj_status("ongoing-request=\"true\"").expect("in-progress form parses"),
2885            RestoreStatus {
2886                is_restore_in_progress: Some(true),
2887                restore_expiry_date: None,
2888            }
2889        );
2890    }
2891
2892    /// A restored object migrated from MinIO must still be recognised as
2893    /// on-disk: `is_restored_object_on_disk` fails open, and
2894    /// `MetaObject::uses_data_dir` uses it to decide whether a data dir is
2895    /// live, so a parse failure here can make a live data dir look reclaimable.
2896    #[test]
2897    fn minio_restored_object_is_recognised_as_on_disk() {
2898        let mut meta = HashMap::new();
2899        meta.insert(
2900            metadata_keys::RESTORE.to_string(),
2901            "ongoing-request=\"false\", expiry-date=\"Fri, 01 Jan 9999 00:00:00 GMT\"".to_string(),
2902        );
2903        assert!(is_restored_object_on_disk(&meta));
2904    }
2905
2906    /// Widening the accepted formats must not weaken malformed-input
2907    /// rejection: an unparseable or non-GMT-shaped expiry still fails closed.
2908    #[test]
2909    fn rejects_malformed_restore_expiry() {
2910        for header in [
2911            "ongoing-request=\"false\", expiry-date=\"not-a-date\"",
2912            "ongoing-request=\"false\", expiry-date=\"Wed, 01 Jan 2025 10:20:30\"",
2913            "ongoing-request=\"false\", expiry-date=\"01 Jan 2025 10:20:30 GMT\"",
2914            "ongoing-request=\"false\"",
2915            "ongoing-request=\"false\", expires=\"2025-01-01T10:20:30Z\"",
2916        ] {
2917            assert!(parse_restore_obj_status(header).is_err(), "{header} must be rejected");
2918        }
2919    }
2920}
rustfs_filemeta/fileinfo.rs

rustfs_filemeta/
fileinfo.rs
