chainfold 0.2.0

//! Fixed 64-byte manifest slot: version, cursor, snapshot id, CRC32C trailer.

use crate::{
    position::Position,
    snapshot::crc32c,
};

/// Byte width of a little-endian u64 field.
const U64_LEN: usize = 8;
/// Byte width of the cursor-present flag.
const FLAG_LEN: usize = 1;
/// Byte width of the CRC32C trailer.
const CRC_LEN: usize = 4;

const VERSION_OFFSET: usize = 0;
const CURSOR_FLAG_OFFSET: usize = VERSION_OFFSET + U64_LEN;
const CURSOR_BLOCK_OFFSET: usize = CURSOR_FLAG_OFFSET + FLAG_LEN;
const CURSOR_LOG_INDEX_OFFSET: usize = CURSOR_BLOCK_OFFSET + U64_LEN;
const SNAPSHOT_ID_OFFSET: usize = CURSOR_LOG_INDEX_OFFSET + U64_LEN;
/// Byte length of the CRC-covered region: every field up to the zero padding.
const CRC_COVERED_LEN: usize = 60;
const CRC_OFFSET: usize = CRC_COVERED_LEN;

/// Byte length of one encoded manifest slot.
pub(crate) const SLOT_SIZE: usize = CRC_OFFSET + CRC_LEN;

/// Byte offset between the two slots: one page, so they never share a failure domain
pub(crate) const SLOT_STRIDE: usize = 4096;

/// Byte length of the whole manifest: the second slot one stride in.
pub(crate) const MANIFEST_SIZE: usize = SLOT_STRIDE + SLOT_SIZE;
/// Manifest file name within a store directory; read by the snapshot store.
pub(crate) const MANIFEST_FILE: &str = "manifest";

/// One manifest slot: durability version, cursor position, and active snapshot id.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct SlotRecord {
    pub(crate) version: u64,
    pub(crate) cursor: Option<Position>,
    pub(crate) snapshot_id: u64,
}

/// Writes a little-endian u64 field at a fixed offset within a slot.
#[inline]
fn put_u64(bytes: &mut [u8; SLOT_SIZE], offset: usize, value: u64) {
    bytes[offset..offset + U64_LEN].copy_from_slice(&value.to_le_bytes());
}

/// Reads a little-endian u64 field at a fixed offset within a full-size slot.
#[inline]
fn get_u64(bytes: &[u8], offset: usize) -> u64 {
    let field: [u8; U64_LEN] = bytes[offset..offset + U64_LEN]
        .try_into()
        .expect("slot is fixed size and every u64 field fits inside it");
    u64::from_le_bytes(field)
}

/// Encodes a slot into its fixed layout with a CRC32C trailer over bytes `0..60`.
pub(crate) fn encode_slot(record: &SlotRecord) -> [u8; SLOT_SIZE] {
    let mut bytes = [0u8; SLOT_SIZE];
    put_u64(&mut bytes, VERSION_OFFSET, record.version);
    bytes[CURSOR_FLAG_OFFSET] = u8::from(record.cursor.is_some());
    put_u64(
        &mut bytes,
        CURSOR_BLOCK_OFFSET,
        record.cursor.map_or(0, |pos| pos.block),
    );
    put_u64(
        &mut bytes,
        CURSOR_LOG_INDEX_OFFSET,
        record.cursor.map_or(0, |pos| pos.log_index),
    );
    put_u64(&mut bytes, SNAPSHOT_ID_OFFSET, record.snapshot_id);
    let crc = crc32c(&bytes[..CRC_COVERED_LEN]);
    bytes[CRC_OFFSET..CRC_OFFSET + CRC_LEN].copy_from_slice(&crc.to_le_bytes());
    bytes
}

/// Decodes a slot, returning None on bad length, bad CRC, or an invalid cursor flag.
pub(crate) fn decode_slot(bytes: &[u8]) -> Option<SlotRecord> {
    if bytes.len() != SLOT_SIZE {
        return None;
    }
    let stored: [u8; CRC_LEN] = bytes[CRC_OFFSET..CRC_OFFSET + CRC_LEN]
        .try_into()
        .expect("slice length matches the CRC field width");
    if crc32c(&bytes[..CRC_COVERED_LEN]) != u32::from_le_bytes(stored) {
        return None;
    }
    let cursor = match bytes[CURSOR_FLAG_OFFSET] {
        0 => None,
        1 => Some(Position::new(
            get_u64(bytes, CURSOR_BLOCK_OFFSET),
            get_u64(bytes, CURSOR_LOG_INDEX_OFFSET),
        )),
        _ => return None,
    };
    Some(SlotRecord {
        version: get_u64(bytes, VERSION_OFFSET),
        cursor,
        snapshot_id: get_u64(bytes, SNAPSHOT_ID_OFFSET),
    })
}

#[cfg(test)]
mod tests {
    use super::{
        SLOT_SIZE,
        SlotRecord,
        decode_slot,
        encode_slot,
    };
    use crate::position::Position;

    #[test]
    fn slot_round_trips_through_encode_decode() {
        // given a record with a cursor and a record without one
        let with_cursor = SlotRecord {
            version: 7,
            cursor: Some(Position::new(100, 3)),
            snapshot_id: 12,
        };
        let without_cursor = SlotRecord {
            version: 1,
            cursor: None,
            snapshot_id: 0,
        };
        // when encoded and decoded
        let decoded_with = decode_slot(&encode_slot(&with_cursor));
        let decoded_without = decode_slot(&encode_slot(&without_cursor));
        // then both round trip equal
        assert_eq!(decoded_with, Some(with_cursor));
        assert_eq!(decoded_without, Some(without_cursor));
    }

    #[test]
    fn slot_rejects_torn_bytes() {
        // given an encoded slot
        let record = SlotRecord {
            version: 3,
            cursor: Some(Position::new(9, 1)),
            snapshot_id: 5,
        };
        let bytes = encode_slot(&record);
        // when any single byte is flipped
        for index in 0..SLOT_SIZE {
            let mut torn = bytes;
            torn[index] = !torn[index];
            // then decode returns None
            assert_eq!(
                decode_slot(&torn),
                None,
                "byte {index} should invalidate the slot"
            );
        }
    }

    #[test]
    fn slot_rejects_wrong_length() {
        // given 63 bytes
        let short = [0u8; SLOT_SIZE - 1];
        // when decoded
        let result = decode_slot(&short);
        // then None
        assert_eq!(result, None);
    }
}