hyperfile 0.12.0

//! What a compaction beside the writer leaves on the inode for the writer.
//!
//! A compactor does not publish a map. It writes copies of blocks, and the map
//! that points at them, under a number already published; then it attaches the
//! result to the stored inode: the inode as it is, with
//! [`InodeRaw::FLAG_COMPACTION_PENDING`] set, and in the inode object's metadata
//! the root of that map and what the writer needs to take it in. The writer takes
//! it in before its next publish of data. See `docs/compaction.md`.
//!
//! [`InodeRaw::FLAG_COMPACTION_PENDING`]: crate::ondisk::InodeRaw::FLAG_COMPACTION_PENDING

use std::io::{Error, ErrorKind, Result};
use crate::Cno;
use crate::ondisk::BMapRawType;

/// The metadata key on the inode object, without the `x-amz-meta-` prefix.
pub const METADATA_KEY: &str = "hf-compaction";

const VERSION: &str = "v1";

/// A compaction attached to the stored inode.
#[derive(Clone, PartialEq, Eq)]
pub struct PendingCompaction {
    /// N0: the checkpoint number of the version the compaction was built on.
    /// A block pointer of the writer's map at a number up to this one, in a node
    /// unchanged since, is as it was in that version.
    pub base: Cno,
    /// `InodeRaw::fingerprint` of that version, for checks and diagnostics.
    pub base_fingerprint: u64,
    /// The parts under `base` the compaction's objects are in, `[p0, p1)`.
    pub parts: (u16, u16),
    /// How many blocks it moved.
    pub moved: u64,
    /// The root of its map: that version's, with the moved blocks at their copies.
    pub root: BMapRawType,
}

impl std::fmt::Debug for PendingCompaction {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "PendingCompaction {{ base: {}, base_fingerprint: {:#018x}, parts: {}..{}, moved: {} }}",
            self.base, self.base_fingerprint, self.parts.0, self.parts.1, self.moved)
    }
}

impl PendingCompaction {
    /// The metadata value: `v1;base=…;fp=…;parts=p0-p1;moved=…;root=…`, ASCII.
    pub fn encode(&self) -> String {
        let root: String = self.root.iter().map(|b| format!("{:02x}", b)).collect();
        format!("{};base={};fp={:016x};parts={}-{};moved={};root={}",
            VERSION, self.base, self.base_fingerprint, self.parts.0, self.parts.1, self.moved, root)
    }

    /// The compaction a metadata value describes, refused with `InvalidData`
    /// unless every field is there and well formed.
    pub fn parse(value: &str) -> Result<Self> {
        let bad = |why: &str| Error::new(ErrorKind::InvalidData,
            format!("compaction metadata {:?}: {}", value, why));
        let mut fields = value.split(';');
        if fields.next() != Some(VERSION) {
            return Err(bad("not a version this build reads"));
        }
        let mut get = |name: &str| -> Result<&str> {
            let f = fields.next().ok_or_else(|| bad(&format!("no {}", name)))?;
            f.strip_prefix(name).and_then(|f| f.strip_prefix('='))
                .ok_or_else(|| bad(&format!("{} expected where {:?} is", name, f)))
        };
        let base: Cno = get("base")?.parse().map_err(|_| bad("base is not a number"))?;
        let fp = get("fp")?;
        let base_fingerprint = u64::from_str_radix(fp, 16).map_err(|_| bad("fp is not hexadecimal"))?;
        let (p0, p1) = get("parts")?.split_once('-').ok_or_else(|| bad("parts is not p0-p1"))?;
        let p0: u16 = p0.parse().map_err(|_| bad("p0 is not a part"))?;
        let p1: u16 = p1.parse().map_err(|_| bad("p1 is not a part"))?;
        let moved: u64 = get("moved")?.parse().map_err(|_| bad("moved is not a number"))?;
        let hex = get("root")?;
        let mut root = [0u8; std::mem::size_of::<BMapRawType>()];
        if hex.len() != root.len() * 2 || !hex.is_ascii() {
            return Err(bad("root is not one map root long"));
        }
        for (i, b) in root.iter_mut().enumerate() {
            *b = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).map_err(|_| bad("root is not hexadecimal"))?;
        }
        if fields.next().is_some() {
            return Err(bad("more fields than this version has"));
        }
        if base == 0 || p0 == 0 || p0 >= p1 || p1 as usize > crate::segment::Segment::MAX_PARTS {
            return Err(bad("base, or the part range, is not one a compaction writes"));
        }
        Ok(Self { base, base_fingerprint, parts: (p0, p1), moved, root })
    }
}

/// Refuse an inode write whose bytes and metadata disagree about a compaction:
/// [`crate::ondisk::InodeRaw::FLAG_COMPACTION_PENDING`] set exactly when one is
/// written with it, and that one built on a number not past the inode's. Every
/// inode write goes through this, so a publish of a new number that kept the bit,
/// or a write that dropped the metadata and not the bit, fails rather than lands.
pub(crate) fn check_inode_write(buf: &[u8], compaction: Option<&PendingCompaction>) -> Result<()> {
    if buf.len() != std::mem::size_of::<crate::ondisk::InodeRaw>() {
        return Ok(());
    }
    let raw = crate::ondisk::InodeRaw::from_u8_slice(buf);
    let bad = |why: String| Err(Error::new(ErrorKind::InvalidInput, why));
    match compaction {
        None if raw.is_compaction_pending() =>
            bad("an inode write sets the attached-compaction bit without the compaction's metadata".into()),
        Some(_) if !raw.is_compaction_pending() =>
            bad("an inode write carries a compaction's metadata without the attached-compaction bit".into()),
        Some(c) if c.base > raw.i_last_cno =>
            bad(format!("a compaction built on checkpoint {} attached to checkpoint {}", c.base, raw.i_last_cno)),
        _ => Ok(()),
    }
}

/// What the stored inode object says of a compaction.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub enum StoredCompaction {
    /// Nothing attached.
    #[default]
    None,
    /// One is attached.
    Pending(PendingCompaction),
    /// Metadata under [`METADATA_KEY`] that does not parse. The writer takes no
    /// compaction in from it, and drops it at its next publish of data; the
    /// cleaner refuses to prune, since it cannot tell what the compaction holds.
    Unreadable(String),
}

impl StoredCompaction {
    /// From the inode object's metadata value, if it has one.
    pub fn from_metadata(value: Option<&str>) -> Self {
        match value {
            None => Self::None,
            Some(v) => match PendingCompaction::parse(v) {
                Ok(p) => Self::Pending(p),
                Err(e) => Self::Unreadable(e.to_string()),
            },
        }
    }

    pub fn pending(&self) -> Option<&PendingCompaction> {
        match self {
            Self::Pending(p) => Some(p),
            _ => None,
        }
    }
}

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

    fn sample() -> PendingCompaction {
        let mut root = [0u8; 56];
        for (i, b) in root.iter_mut().enumerate() { *b = (i * 37) as u8; }
        PendingCompaction { base: 10, base_fingerprint: 0xdead_beef_0123_4567, parts: (3, 9), moved: 4096, root }
    }

    #[test]
    fn round_trip() {
        let p = sample();
        let v = p.encode();
        assert!(v.is_ascii() && v.len() < 2048, "{}", v);
        assert!(v.starts_with("v1;base=10;fp=deadbeef01234567;parts=3-9;moved=4096;root="), "{}", v);
        assert_eq!(PendingCompaction::parse(&v).unwrap(), p);
        assert_eq!(StoredCompaction::from_metadata(Some(&v)), StoredCompaction::Pending(p));
        assert_eq!(StoredCompaction::from_metadata(None), StoredCompaction::None);
    }

    #[test]
    fn an_inode_write_agrees_with_its_metadata() {
        let mut raw = crate::ondisk::InodeRaw::default();
        raw.i_last_cno = 12;
        let p = sample();
        assert!(check_inode_write(raw.as_u8_slice(), None).is_ok());
        assert_eq!(check_inode_write(raw.as_u8_slice(), Some(&p)).unwrap_err().kind(), ErrorKind::InvalidInput);
        raw.set_compaction_pending(true);
        assert_eq!(check_inode_write(raw.as_u8_slice(), None).unwrap_err().kind(), ErrorKind::InvalidInput);
        assert!(check_inode_write(raw.as_u8_slice(), Some(&p)).is_ok());
        raw.i_last_cno = 9;
        assert_eq!(check_inode_write(raw.as_u8_slice(), Some(&p)).unwrap_err().kind(), ErrorKind::InvalidInput);
    }

    #[test]
    fn what_is_refused() {
        let v = sample().encode();
        let root = v.split("root=").nth(1).unwrap().to_string();
        let bad = [
            v.replacen("v1", "v2", 1),
            v.replacen("base=10", "base=x", 1),
            v.replacen("base=10", "base=0", 1),
            v.replacen(";fp=", ";fq=", 1),
            v.replacen("parts=3-9", "parts=9-3", 1),
            v.replacen("parts=3-9", "parts=0-9", 1),
            v.replacen("parts=3-9", "parts=3-20000", 1),
            v.replacen("parts=3-9", "parts=3", 1),
            v.replacen(";moved=4096", "", 1),
            v.replacen(&root, &root[2..], 1),
            v.replacen(&root, &format!("zz{}", &root[2..]), 1),
            format!("{};extra=1", v),
            String::new(),
        ];
        for b in bad {
            let e = PendingCompaction::parse(&b).expect_err(&b);
            assert_eq!(e.kind(), ErrorKind::InvalidData, "{}", b);
            assert!(matches!(StoredCompaction::from_metadata(Some(&b)), StoredCompaction::Unreadable(_)), "{}", b);
        }
    }
}