hyperfile 0.11.2

//! Objects laid out tail first, for [`BlockPtrFormat::StreamSegment`].
//!
//! ```text
//! [data blocks][meta blocks][summary entries][header]
//!  n × data     m × meta     n × 16           256 B
//! ```
//!
//! The order is what each part depends on. A data block's place is fixed when it
//! is placed — `k × data block size` — so it can be written before anything after
//! it is known. The index nodes point at data blocks, so they come after; the
//! summary lists the data blocks; the header counts everything, names the object
//! and carries the inode, so it comes last.
//!
//! Every object describes itself: its data, the index nodes over them, a summary
//! naming each block, and a header saying which object of which checkpoint it is.
//! A reader finds the header in the last [`HEADER_SIZE`] bytes and everything else
//! from it; one read of the object's end usually returns the header, the summary
//! and the index nodes together.
//!
//! The object begins with data, so nothing at its start says what it is: a reader
//! that does not know the layout looks at the end first. A header is taken only if
//! its magic, version and checksum hold, its counts account for exactly the
//! object's length, and it names the object it was read from — so bytes at the end
//! of an object in another layout are not mistaken for one.
//!
//! [`BlockPtrFormat::StreamSegment`]: crate::meta_format::BlockPtrFormat::StreamSegment

use std::io::{Error, ErrorKind, Result};

use crate::ondisk::{InodeRaw, SegmentBlockEntryRaw, SegmentHeader, SEGMENT_FLAG_PARTIAL};
use crate::segment::{SegmentBlockDesc, SegmentSum};
use crate::{BlockIndex, BlockPtr, SegmentId};

/// The magic a header begins with.
pub const STREAM_HEADER_MAGIC: u32 = u32::from_le_bytes(*b"HFST");
/// The layout version a header records.
pub const STREAM_VERSION: u16 = 1;
/// The header's size: the object's last bytes.
pub const HEADER_SIZE: usize = 256;
/// One summary entry: block index and pointer.
pub const ENTRY_SIZE: usize = std::mem::size_of::<SegmentBlockEntryRaw>();

const INODE_SIZE: usize = std::mem::size_of::<InodeRaw>();
// Field offsets within the header. Bytes not named are reserved and zero.
const H_MAGIC: usize = 0;
const H_VERSION: usize = 4;
const H_FLAGS: usize = 6;
const H_CNO: usize = 8;
const H_PART: usize = 16;
const H_META_SHIFT: usize = 18;
const H_DATA_SHIFT: usize = 19;
const H_NDATA: usize = 20;
const H_NMETA: usize = 24;
const H_BASE: usize = 32;
const H_INODE: usize = 40;
const H_CRC: usize = HEADER_SIZE - 4;
const _: () = assert!(H_INODE + INODE_SIZE <= H_CRC);

/// CRC-32 (IEEE) of `bytes`.
fn crc32(bytes: &[u8]) -> u32 {
    let mut crc = !0u32;
    for b in bytes {
        crc ^= *b as u32;
        for _ in 0..8 {
            crc = (crc >> 1) ^ (0xEDB8_8320 & (crc & 1).wrapping_neg());
        }
    }
    !crc
}

/// The bytes of an object holding `ndata` data blocks and `nmeta` index nodes.
pub fn object_len(ndata: usize, data_block_size: usize, nmeta: usize, meta_block_size: usize) -> usize {
    ndata * data_block_size + nmeta * meta_block_size + ndata * ENTRY_SIZE + HEADER_SIZE
}

/// Where data block `k` of an object starts.
#[inline]
pub fn data_off(k: usize, data_block_size: usize) -> usize {
    k * data_block_size
}

/// What the end of an object says about it.
#[derive(Debug, Clone, Copy)]
pub struct StreamHeader {
    pub version: u16,
    pub flags: u16,
    /// The object this is: a checkpoint and one of its parts.
    pub object: SegmentId,
    pub meta_shift: u8,
    pub data_shift: u8,
    pub ndata: u32,
    pub nmeta: u32,
    /// Fingerprint of the inode the checkpoint was built on; see `SegmentHeader::s_base`.
    pub base: u64,
    pub inode: InodeRaw,
}

impl StreamHeader {
    pub fn new(object: SegmentId, meta_block_size: usize, data_block_size: usize, partial: bool, base: u64) -> Self {
        debug_assert!(object.part_id().is_some(), "a stream object is always a part");
        Self {
            version: STREAM_VERSION,
            flags: if partial { SEGMENT_FLAG_PARTIAL } else { 0 },
            object,
            meta_shift: meta_block_size.trailing_zeros() as u8,
            data_shift: data_block_size.trailing_zeros() as u8,
            ndata: 0,
            nmeta: 0,
            base,
            inode: unsafe { std::mem::MaybeUninit::zeroed().assume_init() },
        }
    }

    pub fn is_partial(&self) -> bool {
        self.flags & SEGMENT_FLAG_PARTIAL != 0
    }

    pub fn write_to(&self, out: &mut [u8]) {
        assert_eq!(out.len(), HEADER_SIZE);
        out.fill(0);
        out[H_MAGIC..H_MAGIC + 4].copy_from_slice(&STREAM_HEADER_MAGIC.to_le_bytes());
        out[H_VERSION..H_VERSION + 2].copy_from_slice(&self.version.to_le_bytes());
        out[H_FLAGS..H_FLAGS + 2].copy_from_slice(&self.flags.to_le_bytes());
        out[H_CNO..H_CNO + 8].copy_from_slice(&self.object.as_cno().to_le_bytes());
        out[H_PART..H_PART + 2].copy_from_slice(&self.object.part_id().expect("a part").to_le_bytes());
        out[H_META_SHIFT] = self.meta_shift;
        out[H_DATA_SHIFT] = self.data_shift;
        out[H_NDATA..H_NDATA + 4].copy_from_slice(&self.ndata.to_le_bytes());
        out[H_NMETA..H_NMETA + 4].copy_from_slice(&self.nmeta.to_le_bytes());
        out[H_BASE..H_BASE + 8].copy_from_slice(&self.base.to_le_bytes());
        out[H_INODE..H_INODE + INODE_SIZE].copy_from_slice(self.inode.as_u8_slice());
        let crc = crc32(&out[..H_CRC]);
        out[H_CRC..].copy_from_slice(&crc.to_le_bytes());
    }

    /// The header in `bytes`, if they are one: magic, version and checksum.
    pub fn read_from(bytes: &[u8]) -> Result<Self> {
        let bad = |why: String| Error::new(ErrorKind::InvalidData, why);
        if bytes.len() != HEADER_SIZE {
            return Err(bad(format!("a stream header is {} bytes, not {}", HEADER_SIZE, bytes.len())));
        }
        let u16_at = |o: usize| u16::from_le_bytes(bytes[o..o + 2].try_into().unwrap());
        let u32_at = |o: usize| u32::from_le_bytes(bytes[o..o + 4].try_into().unwrap());
        let u64_at = |o: usize| u64::from_le_bytes(bytes[o..o + 8].try_into().unwrap());
        if u32_at(H_MAGIC) != STREAM_HEADER_MAGIC {
            return Err(bad(format!("not a stream header (magic {:#x})", u32_at(H_MAGIC))));
        }
        if u32_at(H_CRC) != crc32(&bytes[..H_CRC]) {
            return Err(bad("a stream header whose checksum does not match".to_string()));
        }
        let version = u16_at(H_VERSION);
        if version != STREAM_VERSION {
            return Err(Error::new(ErrorKind::Unsupported, format!(
                "a stream header of layout version {}; this build reads {}", version, STREAM_VERSION)));
        }
        let (meta_shift, data_shift) = (bytes[H_META_SHIFT], bytes[H_DATA_SHIFT]);
        if !(12..=30).contains(&meta_shift) || !(12..=30).contains(&data_shift) {
            return Err(bad(format!("a stream header with block shifts {} and {}", meta_shift, data_shift)));
        }
        let mut inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        inode.as_mut_u8_slice().copy_from_slice(&bytes[H_INODE..H_INODE + INODE_SIZE]);
        Ok(Self {
            version,
            flags: u16_at(H_FLAGS),
            object: SegmentId::new_from_cno(u64_at(H_CNO)).at_part(u16_at(H_PART)),
            meta_shift,
            data_shift,
            ndata: u32_at(H_NDATA),
            nmeta: u32_at(H_NMETA),
            base: u64_at(H_BASE),
            inode,
        })
    }

    /// Bytes from the start of the index nodes to the end: what a reader needs
    /// beyond the data.
    pub fn tail_len(&self) -> usize {
        ((self.nmeta as usize) << self.meta_shift) + self.ndata as usize * ENTRY_SIZE + HEADER_SIZE
    }

    /// The bytes of the object this header describes.
    pub fn object_len(&self) -> usize {
        ((self.ndata as usize) << self.data_shift) + self.tail_len()
    }

    /// As the header the older layout begins with, for code that works on a
    /// summary whatever the layout.
    pub fn as_segment_header(&self) -> SegmentHeader {
        let mut h = SegmentHeader::new();
        h.s_magic = STREAM_HEADER_MAGIC;
        h.s_bytes = (HEADER_SIZE + self.ndata as usize * ENTRY_SIZE) as u32;
        h.s_flags = self.flags;
        h.s_meta_blk_shift = self.meta_shift;
        h.s_data_blk_shift = self.data_shift;
        h.s_base = self.base;
        h.s_inode = self.inode;
        h.s_cno = self.object.as_cno();
        h.s_nmetablk = self.nmeta;
        h.s_ndatablk = self.ndata;
        h
    }
}

/// Where everything in an object laid out tail first is.
#[derive(Debug, Clone)]
pub struct StreamLayout {
    pub hdr: StreamHeader,
    /// The object's length.
    pub total: usize,
    pub meta_off: usize,
    pub summary_off: usize,
    pub header_off: usize,
}

impl StreamLayout {
    /// The layout of an object of `total` bytes whose last [`HEADER_SIZE`] bytes
    /// are `header`, read as `object`. Refused unless the header holds, accounts
    /// for exactly `total` bytes, and names `object`.
    pub fn from_header(total: usize, header: &[u8], object: SegmentId) -> Result<Self> {
        let hdr = StreamHeader::read_from(header)?;
        if hdr.object_len() != total {
            return Err(Error::new(ErrorKind::InvalidData, format!(
                "a stream header counts {} data and {} meta blocks, {} bytes, in an object of {}",
                hdr.ndata, hdr.nmeta, hdr.object_len(), total)));
        }
        if hdr.object != object {
            return Err(Error::new(ErrorKind::InvalidData, format!(
                "a stream header naming {} in the object {}", hdr.object, object)));
        }
        let meta_off = (hdr.ndata as usize) << hdr.data_shift;
        let summary_off = meta_off + ((hdr.nmeta as usize) << hdr.meta_shift);
        Ok(Self { hdr, total, meta_off, summary_off, header_off: total - HEADER_SIZE })
    }

    pub fn meta_block_size(&self) -> usize {
        1 << self.hdr.meta_shift
    }

    pub fn nmeta(&self) -> usize {
        self.hdr.nmeta as usize
    }

    /// The summary, from `entries`: the object's bytes `summary_off..header_off`.
    pub fn summary(&self, entries: &[u8]) -> SegmentSum {
        assert_eq!(entries.len(), self.header_off - self.summary_off);
        let blocks = entries.chunks(ENTRY_SIZE).map(|e| {
            let raw = SegmentBlockEntryRaw::read_from(e);
            SegmentBlockDesc { blkidx: raw.e_blkidx, blkptr: raw.e_blkptr }
        }).collect();
        SegmentSum { hdr: self.hdr.as_segment_header(), blocks }
    }
}

/// An object's layout and summary from its last bytes: `tail` is the end of the
/// object `object`, `total` bytes long. `Ok(Err(n))` when `tail` is too short and
/// the last `n` bytes are needed.
pub fn parse_tail(object: SegmentId, total: usize, tail: &[u8])
    -> Result<std::result::Result<(StreamLayout, SegmentSum), usize>>
{
    if tail.len() < HEADER_SIZE {
        return Ok(Err(HEADER_SIZE));
    }
    let layout = StreamLayout::from_header(total, &tail[tail.len() - HEADER_SIZE..], object)?;
    let need = total - layout.summary_off;
    if tail.len() < need {
        return Ok(Err(need));
    }
    let base = total - tail.len();
    let sum = layout.summary(&tail[layout.summary_off - base..layout.header_off - base]);
    Ok(Ok((layout, sum)))
}

/// Builds the end of one object: the summary entries and the header. The data and
/// the index nodes are appended by the caller in the layout's order.
#[derive(Debug)]
pub struct TailBuilder {
    pub hdr: StreamHeader,
    entries: Vec<(BlockIndex, BlockPtr)>,
}

impl TailBuilder {
    pub fn new(object: SegmentId, meta_block_size: usize, data_block_size: usize, partial: bool, base: u64) -> Self {
        Self { hdr: StreamHeader::new(object, meta_block_size, data_block_size, partial, base), entries: Vec::new() }
    }

    pub fn data(&mut self, blk_idx: BlockIndex, blk_ptr: BlockPtr) {
        self.entries.push((blk_idx, blk_ptr));
        self.hdr.ndata += 1;
    }

    pub fn meta(&mut self) {
        self.hdr.nmeta += 1;
    }

    pub fn ndata(&self) -> usize {
        self.entries.len()
    }

    /// The summary entries followed by the header carrying `inode`.
    pub fn finish(mut self, inode: &InodeRaw) -> Vec<u8> {
        self.hdr.inode = *inode;
        let mut out = vec![0u8; self.entries.len() * ENTRY_SIZE + HEADER_SIZE];
        for (k, (blkidx, blkptr)) in self.entries.iter().enumerate() {
            SegmentBlockEntryRaw { e_blkidx: *blkidx, e_blkptr: *blkptr }
                .write_to(&mut out[k * ENTRY_SIZE..(k + 1) * ENTRY_SIZE]);
        }
        let at = out.len() - HEADER_SIZE;
        self.hdr.write_to(&mut out[at..]);
        out
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const BS: usize = 4096;

    fn id() -> SegmentId {
        SegmentId::new_from_cno(7).at_part(3)
    }

    fn object(ndata: usize, nmeta: usize, partial: bool) -> Vec<u8> {
        let mut obj = Vec::new();
        let mut t = TailBuilder::new(id(), BS, BS, partial, 0xba5e);
        for k in 0..ndata {
            assert_eq!(obj.len(), data_off(k, BS));
            obj.extend(std::iter::repeat(k as u8).take(BS));
            t.data(100 + k as u64, 0x4000_0000_0000_0000 | k as u64);
        }
        for j in 0..nmeta {
            obj.extend(std::iter::repeat(0xA0 | j as u8).take(BS));
            t.meta();
        }
        let mut inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        inode.i_last_cno = 7;
        obj.extend(t.finish(&inode));
        assert_eq!(obj.len(), object_len(ndata, BS, nmeta, BS));
        obj
    }

    #[test]
    fn an_object_parses_from_its_end() {
        for (ndata, nmeta) in [(0, 0), (1, 0), (0, 1), (5, 2), (300, 3)] {
            let obj = object(ndata, nmeta, true);
            let (layout, sum) = parse_tail(id(), obj.len(), &obj).unwrap().unwrap();
            assert_eq!(layout.meta_off, data_off(ndata, BS));
            assert_eq!(layout.nmeta(), nmeta);
            assert_eq!(layout.hdr.object, id());
            assert!(layout.hdr.is_partial());
            assert_eq!(layout.hdr.base, 0xba5e);
            assert_eq!(layout.hdr.inode.i_last_cno, 7);
            assert_eq!(sum.hdr.s_cno, 7);
            assert_eq!(sum.hdr.s_flags & SEGMENT_FLAG_PARTIAL, SEGMENT_FLAG_PARTIAL);
            assert_eq!(sum.blocks.len(), ndata);
            for (k, b) in sum.blocks.iter().enumerate() {
                assert_eq!(b.blkidx, 100 + k as u64);
                assert_eq!(&obj[data_off(k, BS)..data_off(k, BS) + BS], &vec![k as u8; BS][..]);
            }
            for j in 0..nmeta {
                assert_eq!(obj[layout.meta_off + j * BS], 0xA0 | j as u8);
            }
        }
    }

    #[test]
    fn a_short_tail_says_how_much_is_needed() {
        let obj = object(300, 3, false);
        let needed = |tail: &[u8]| match parse_tail(id(), obj.len(), tail).unwrap() { Err(n) => n, Ok(_) => 0 };
        assert_eq!(needed(&obj[obj.len() - 100..]), HEADER_SIZE);
        let need = needed(&obj[obj.len() - HEADER_SIZE..]);
        assert_eq!(need, 300 * ENTRY_SIZE + HEADER_SIZE);
        assert!(parse_tail(id(), obj.len(), &obj[obj.len() - need..]).unwrap().is_ok());
    }

    #[test]
    fn a_header_that_does_not_hold_is_refused() {
        let obj = object(5, 2, false);
        let n = obj.len();
        // Another length, another object, a flipped bit.
        assert!(parse_tail(id(), n - 1, &obj[1..]).is_err());
        assert!(parse_tail(SegmentId::new_from_cno(7).at_part(4), n, &obj).is_err());
        assert!(parse_tail(SegmentId::new_from_cno(8).at_part(3), n, &obj).is_err());
        let mut flipped = obj.clone();
        flipped[n - HEADER_SIZE + H_NDATA] ^= 1;
        assert!(parse_tail(id(), n, &flipped).is_err());
        // The older layout's header at an object's end, and plain data.
        let mut old = vec![0u8; HEADER_SIZE];
        SegmentHeader::new().write_to(&mut old[..SegmentHeader::size()]);
        assert!(StreamHeader::read_from(&old).is_err());
        assert!(StreamHeader::read_from(&vec![0x48u8; HEADER_SIZE]).is_err());
    }

    #[test]
    fn a_later_layout_version_is_unsupported_not_garbage() {
        let obj = object(1, 1, false);
        let n = obj.len();
        let mut h = obj[n - HEADER_SIZE..].to_vec();
        h[H_VERSION..H_VERSION + 2].copy_from_slice(&2u16.to_le_bytes());
        let crc = crc32(&h[..H_CRC]);
        h[H_CRC..].copy_from_slice(&crc.to_le_bytes());
        assert_eq!(StreamHeader::read_from(&h).unwrap_err().kind(), ErrorKind::Unsupported);
    }
}