rustfs_filemeta/filemeta/
codec.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 super::*;
16
17impl FileMeta {
18    pub fn is_xl2_v1_format(buf: &[u8]) -> bool {
19        !matches!(Self::check_xl2_v1(buf), Err(_e))
20    }
21
22    #[hotpath::measure(impl_type = "FileMeta")]
23    pub fn load(buf: &[u8]) -> Result<FileMeta> {
24        let mut xl = FileMeta::default();
25        xl.unmarshal_msg(buf)?;
26        Ok(xl)
27    }
28
29    /// Read (major, minor, header_ver, meta_ver) from xl.meta without full parse.
30    /// Returns Err if format is not feat (e.g. legacy  uses rmp_serde in meta block).
31    pub fn read_format_versions(buf: &[u8]) -> Result<(u16, u16, u8, u8)> {
32        let (buf, major, minor) = Self::check_xl2_v1(buf)?;
33        if buf.len() < 5 {
34            return Err(Error::FileCorrupt);
35        }
36        let (mut size_buf, buf) = buf.split_at(5);
37        let bin_len = rmp::decode::read_bin_len(&mut size_buf).map_err(|_| Error::FileCorrupt)?;
38        if buf.len() < bin_len as usize {
39            return Err(Error::FileCorrupt);
40        }
41        let (meta, _) = buf.split_at(bin_len as usize);
42        let (_, header_ver, meta_ver, _) = Self::decode_xl_headers(meta)?;
43        Ok((major, minor, header_ver, meta_ver))
44    }
45
46    pub fn check_xl2_v1(buf: &[u8]) -> Result<(&[u8], u16, u16)> {
47        // A file too short to hold the XL2 magic, or one that carries the
48        // wrong magic, is not merely unreadable — it is affirmative evidence
49        // of a torn or foreign write. Classify it as FileCorrupt so quorum
50        // and listing code can distinguish deterministic damage from
51        // transient IO faults (issue #5716).
52        if buf.len() < 8 {
53            return Err(Error::FileCorrupt);
54        }
55
56        if buf[0..4] != XL_FILE_HEADER {
57            return Err(Error::FileCorrupt);
58        }
59
60        let major = byteorder::LittleEndian::read_u16(&buf[4..6]);
61        let minor = byteorder::LittleEndian::read_u16(&buf[6..8]);
62        if major > XL_FILE_VERSION_MAJOR {
63            return Err(Error::other("xl file version err"));
64        }
65
66        Ok((&buf[8..], major, minor))
67    }
68
69    // Returns (meta, inline_data)
70    pub fn is_indexed_meta(buf: &[u8]) -> Result<(&[u8], &[u8])> {
71        let (buf, major, minor) = Self::check_xl2_v1(buf)?;
72        if major != 1 || minor < 3 {
73            return Ok((&[], &[]));
74        }
75
76        if buf.len() < 5 {
77            return Err(Error::FileCorrupt);
78        }
79
80        let (mut size_buf, buf) = buf.split_at(5);
81
82        // Get meta data, buf = crc + data
83        let bin_len = rmp::decode::read_bin_len(&mut size_buf).map_err(|_| Error::FileCorrupt)?;
84
85        if buf.len() < bin_len as usize {
86            return Err(Error::FileCorrupt);
87        }
88        let (meta, buf) = buf.split_at(bin_len as usize);
89
90        if buf.len() < 5 {
91            return Err(Error::FileCorrupt);
92        }
93        let (mut crc_buf, inline_data) = buf.split_at(5);
94
95        // crc check
96        let crc = rmp::decode::read_u32(&mut crc_buf).map_err(|_| Error::FileCorrupt)?;
97        let meta_crc = xxh64::xxh64(meta, XXHASH_SEED) as u32;
98
99        if crc != meta_crc {
100            error!(
101                event = "filemeta_xl_crc_mismatch",
102                component = "filemeta",
103                expected_crc = meta_crc,
104                actual_crc = crc,
105                "xl.meta payload failed its CRC check"
106            );
107            // Error::FileCorrupt, not a generic error, for the same reason
108            // check_xl2_v1 classifies a bad magic as FileCorrupt: heal
109            // classification (should_heal_object_on_disk) recognises
110            // corruption only by the DiskError::FileCorrupt variant this
111            // converts to. As a generic error the drive is skipped,
112            // heal_object reports ok, and on-disk bitrot is never repaired.
113            return Err(Error::FileCorrupt);
114        }
115
116        Ok((meta, inline_data))
117    }
118
119    // Fixed u32
120    pub fn read_bytes_header(buf: &[u8]) -> Result<(u32, &[u8])> {
121        if buf.len() < 5 {
122            return Err(Error::FileCorrupt);
123        }
124
125        let (mut size_buf, remaining) = buf.split_at(5);
126
127        // Get meta data, buf = crc + data
128        let bin_len = rmp::decode::read_bin_len(&mut size_buf).map_err(|_| Error::FileCorrupt)?;
129
130        Ok((bin_len, remaining))
131    }
132
133    #[hotpath::measure(impl_type = "FileMeta")]
134    pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result<u64> {
135        let i = buf.len() as u64;
136
137        // check version, buf = buf[8..]
138        let (buf, _, _) = Self::check_xl2_v1(buf)?;
139
140        // These bytes have already been read. Invalid framing is deterministic
141        // metadata damage; preserve FileCorrupt so quorum-backed heal can repair it.
142        if buf.len() < 5 {
143            error!(
144                "insufficient data for metadata length prefix: expected at least 5 bytes, got {}",
145                buf.len()
146            );
147            return Err(Error::FileCorrupt);
148        }
149
150        let (mut size_buf, buf) = buf.split_at(5);
151
152        // Get meta data, buf = crc + data
153        let bin_len = rmp::decode::read_bin_len(&mut size_buf).map_err(|e| {
154            error!("failed to read binary length for metadata: {}", e);
155            Error::FileCorrupt
156        })?;
157
158        if buf.len() < bin_len as usize {
159            error!("insufficient data for metadata: expected {} bytes, got {} bytes", bin_len, buf.len());
160            return Err(Error::FileCorrupt);
161        }
162        let (meta, buf) = buf.split_at(bin_len as usize);
163
164        if buf.len() < 5 {
165            error!("insufficient data for CRC: expected 5 bytes, got {} bytes", buf.len());
166            return Err(Error::FileCorrupt);
167        }
168        let (mut crc_buf, buf) = buf.split_at(5);
169
170        // crc check
171        let crc = rmp::decode::read_u32(&mut crc_buf).map_err(|e| {
172            error!("failed to read CRC value: {}", e);
173            Error::FileCorrupt
174        })?;
175        let meta_crc = xxh64::xxh64(meta, XXHASH_SEED) as u32;
176
177        if crc != meta_crc {
178            error!(
179                event = "filemeta_xl_crc_mismatch",
180                component = "filemeta",
181                expected_crc = meta_crc,
182                actual_crc = crc,
183                "xl.meta payload failed its CRC check"
184            );
185            // See is_indexed_meta: the FileCorrupt variant is what makes heal
186            // classify this drive as needing metadata repair.
187            return Err(Error::FileCorrupt);
188        }
189
190        if !buf.is_empty() {
191            self.data.update(buf);
192            self.data.validate().map_err(|e| {
193                error!("data validation failed: {}", e);
194                e
195            })?;
196        }
197
198        // Parse meta
199        if !meta.is_empty() {
200            let (versions_len, header_ver, meta_ver, meta) = Self::decode_xl_headers(meta).map_err(|e| {
201                error!("failed to decode XL headers: {}", e);
202                e
203            })?;
204
205            // let (_, meta) = meta.split_at(read_size as usize);
206
207            self.meta_ver = meta_ver;
208
209            // `versions_len` is decoded from a potentially corrupted buffer.
210            // Every version contributes at least two msgpack bin headers to
211            // `meta`, so a count larger than the remaining bytes is corrupt;
212            // reject it before sizing any allocation from it (rustfs/rustfs#2715).
213            if versions_len > meta.len() {
214                error!(
215                    "corrupt xl.meta: version count {} exceeds remaining metadata size {}",
216                    versions_len,
217                    meta.len()
218                );
219                return Err(Error::other(format!(
220                    "corrupt xl.meta: version count {versions_len} exceeds metadata size {}",
221                    meta.len()
222                )));
223            }
224
225            self.versions = Vec::with_capacity(versions_len);
226
227            let mut cur: Cursor<&[u8]> = Cursor::new(meta);
228            for _ in 0..versions_len {
229                let bin_len = rmp::decode::read_bin_len(&mut cur).map_err(|e| {
230                    error!("failed to read binary length for version header: {}", e);
231                    Error::other(format!("failed to read binary length for version header: {e}"))
232                })? as usize;
233
234                let remaining = meta.len().saturating_sub(cur.position() as usize);
235                if bin_len > remaining {
236                    error!("corrupt xl.meta: version header length {} exceeds remaining {} bytes", bin_len, remaining);
237                    return Err(Error::other("corrupt xl.meta: version header length exceeds metadata size"));
238                }
239                let mut header_buf = vec![0u8; bin_len];
240
241                cur.read_exact(&mut header_buf)?;
242
243                let mut ver = FileMetaShallowVersion::default();
244                ver.header.unmarshal_v(header_ver, &header_buf).map_err(|e| {
245                    error!("failed to unmarshal version header: {}", e);
246                    e
247                })?;
248
249                let bin_len = rmp::decode::read_bin_len(&mut cur).map_err(|e| {
250                    error!("failed to read binary length for version metadata: {}", e);
251                    Error::other(format!("failed to read binary length for version metadata: {e}"))
252                })? as usize;
253
254                let remaining = meta.len().saturating_sub(cur.position() as usize);
255                if bin_len > remaining {
256                    error!(
257                        "corrupt xl.meta: version metadata length {} exceeds remaining {} bytes",
258                        bin_len, remaining
259                    );
260                    return Err(Error::other("corrupt xl.meta: version metadata length exceeds metadata size"));
261                }
262                let mut ver_meta_buf = vec![0u8; bin_len];
263                cur.read_exact(&mut ver_meta_buf)?;
264
265                ver.meta.extend_from_slice(&ver_meta_buf);
266
267                self.versions.push(ver);
268            }
269        }
270
271        Ok(i)
272    }
273
274    // decode_xl_headers parses meta header, returns (versions count, xl_header_version, xl_meta_version, read data length)
275    fn decode_xl_headers(buf: &[u8]) -> Result<(usize, u8, u8, &[u8])> {
276        let mut cur = Cursor::new(buf);
277
278        let header_ver: u8 = rmp::decode::read_int(&mut cur)?;
279
280        if header_ver > XL_HEADER_VERSION {
281            return Err(Error::other("xl header version invalid"));
282        }
283
284        let meta_ver: u8 = rmp::decode::read_int(&mut cur)?;
285        if meta_ver > XL_META_VERSION {
286            return Err(Error::other("xl meta version invalid"));
287        }
288
289        let versions_len: usize = rmp::decode::read_int(&mut cur)?;
290
291        Ok((versions_len, header_ver, meta_ver, &buf[cur.position() as usize..]))
292    }
293
294    fn decode_versions<F: FnMut(usize, &[u8], &[u8]) -> Result<()>>(buf: &[u8], versions: usize, mut fnc: F) -> Result<()> {
295        let mut cur: Cursor<&[u8]> = Cursor::new(buf);
296
297        for i in 0..versions {
298            let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize;
299            let start = cur.position() as usize;
300            let end = start + bin_len;
301            let header_buf = buf
302                .get(start..end)
303                .ok_or_else(|| Error::other("corrupt xl.meta: version header segment out of range"))?;
304
305            cur.set_position(end as u64);
306
307            let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize;
308            let start = cur.position() as usize;
309            let end = start + bin_len;
310            let ver_meta_buf = buf
311                .get(start..end)
312                .ok_or_else(|| Error::other("corrupt xl.meta: version metadata segment out of range"))?;
313
314            cur.set_position(end as u64);
315
316            if let Err(err) = fnc(i, header_buf, ver_meta_buf) {
317                if err == Error::DoneForNow {
318                    return Ok(());
319                }
320
321                return Err(err);
322            }
323        }
324
325        Ok(())
326    }
327
328    pub fn is_latest_delete_marker(buf: &[u8]) -> bool {
329        let header = Self::decode_xl_headers(buf).ok();
330        if let Some((versions, hdr_v, _meta_v, meta)) = header {
331            if versions == 0 {
332                return false;
333            }
334
335            let mut is_delete_marker = false;
336
337            let _ = Self::decode_versions(meta, versions, |_: usize, hdr: &[u8], _: &[u8]| {
338                let mut header = FileMetaVersionHeader::default();
339                if header.unmarshal_v(hdr_v, hdr).is_err() {
340                    return Err(Error::DoneForNow);
341                }
342
343                is_delete_marker = header.version_type == VersionType::Delete;
344
345                Err(Error::DoneForNow)
346            });
347
348            is_delete_marker
349        } else {
350            false
351        }
352    }
353
354    #[hotpath::measure(impl_type = "FileMeta")]
355    pub fn marshal_msg(&self) -> Result<Vec<u8>> {
356        let mut wr = Vec::new();
357
358        // header
359        wr.write_all(XL_FILE_HEADER.as_slice())?;
360
361        let mut major = [0u8; 2];
362        byteorder::LittleEndian::write_u16(&mut major, XL_FILE_VERSION_MAJOR);
363        wr.write_all(major.as_slice())?;
364
365        let mut minor = [0u8; 2];
366        byteorder::LittleEndian::write_u16(&mut minor, XL_FILE_VERSION_MINOR);
367        wr.write_all(minor.as_slice())?;
368
369        // size bin32 reserved for write_bin_len
370        wr.write_all(&[0xc6, 0, 0, 0, 0])?;
371
372        let offset = wr.len();
373
374        rmp::encode::write_uint(&mut wr, XL_HEADER_VERSION as u64)?;
375        rmp::encode::write_uint(&mut wr, XL_META_VERSION as u64)?;
376
377        rmp::encode::write_sint(&mut wr, self.versions.len() as i64)?;
378
379        for ver in self.versions.iter() {
380            let hmsg = ver.header.marshal_msg()?;
381            rmp::encode::write_bin(&mut wr, &hmsg)?;
382
383            rmp::encode::write_bin(&mut wr, &ver.meta)?;
384        }
385
386        // Update bin length
387        let data_len = wr.len() - offset;
388        byteorder::BigEndian::write_u32(&mut wr[offset - 4..offset], data_len as u32);
389
390        let crc = xxh64::xxh64(&wr[offset..], XXHASH_SEED) as u32;
391        let mut crc_buf = [0u8; 5];
392        crc_buf[0] = 0xce; // u32
393        byteorder::BigEndian::write_u32(&mut crc_buf[1..], crc);
394
395        wr.write_all(&crc_buf)?;
396
397        wr.write_all(self.data.as_slice())?;
398
399        Ok(wr)
400    }
401
402    // pub fn unmarshal(buf: &[u8]) -> Result<Self> {
403    //     let mut s = Self::default();
404    //     s.unmarshal_msg(buf)?;
405    //     Ok(s)
406    //     // let t: FileMeta = rmp_serde::from_slice(buf)?;
407    //     // Ok(t)
408    // }
409
410    // pub fn marshal_msg(&self) -> Result<Vec<u8>> {
411    //     let mut buf = Vec::new();
412
413    //     self.serialize(&mut Serializer::new(&mut buf))?;
414
415    //     Ok(buf)
416    // }
417}
rustfs_filemeta/filemeta/codec.rs

rustfs_filemeta/filemeta/
codec.rs
