lace_sans_io/records/

record.rs

//! Validated records over one retained byte source.

use std::fmt;
use std::hash::{Hash, Hasher};
use std::io;
use std::sync::Arc;

#[cfg(all(target_pointer_width = "64", any(unix, windows)))]
use crate::runtime_support::native_record_arena::ArenaMapping;

use super::file::RecordFile;
use super::metadata::{RecordMetadata, trusted_data_offset, trusted_head_data_offset};
use super::validator::{RecordError, RecordValidator, ValidatedHeader};
use crate::hash::{RecordHash, RecordKind};
use crate::util::MAX_RECORD_METADATA_BYTES;

/// Bytes a trusted file-range load reads first; most prefixes fit, longer
/// ones take one more read of at most the maximum prefix.
const FIRST_PREFIX_READ: usize = 1024;

/// The immutable canonical bytes a record retains.
#[derive(Clone)]
enum RecordSource {
    Memory(Arc<[u8]>),
    /// A committed native arena range. Only an accepted record-log row
    /// constructs it; the mapping holds the shared arena lock.
    #[cfg(all(target_pointer_width = "64", any(unix, windows)))]
    Mmap {
        mmap: Arc<ArenaMapping>,
        offset: usize,
        len: usize,
    },
    /// A range of an immutable record file, read in bounded pieces. Its
    /// canonical prefix is held in memory.
    File {
        file: Arc<RecordFile>,
        offset: u64,
        len: usize,
        prefix: Arc<[u8]>,
    },
}

impl RecordSource {
    fn contiguous(&self) -> Option<&[u8]> {
        Some(match self {
            Self::Memory(bytes) => bytes,
            #[cfg(all(target_pointer_width = "64", any(unix, windows)))]
            Self::Mmap { mmap, offset, len } => {
                // SAFETY: the range was bounds-checked inside a permanent
                // fixed-size mapping when an accepted committed row
                // constructed it. Published ranges are immutable while any
                // mapping owner holds its shared arena lock, which excludes
                // compaction.
                unsafe { std::slice::from_raw_parts(mmap.as_ptr().add(*offset), *len) }
            }
            Self::File { .. } => return None,
        })
    }
}

/// A validated record: verified canonical bytes from one retained source.
///
/// Its metadata (fields, links, `By`/Mark, lengths, containment) is always in
/// memory. The complete bytes are available as one slice only where the
/// source is contiguous; [`Record::read_at`] reads any source in bounded
/// pieces. Clones share the source. Equality and hashing use the record hash.
#[derive(Clone)]
pub struct Record {
    source: RecordSource,
    data_offset: usize,
}

impl fmt::Debug for Record {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("Record")
            .field("hash", &self.hash())
            .field("canonical_len", &self.canonical_len())
            .finish()
    }
}

impl PartialEq for Record {
    fn eq(&self, other: &Self) -> bool {
        self.hash() == other.hash()
    }
}
impl Eq for Record {}
impl Hash for Record {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.hash().hash(state);
    }
}

/// A record could not be materialised.
#[derive(Debug)]
pub enum RecordReadError {
    /// The requested bytes exceed the caller's limit.
    TooLarge { length: usize, limit: usize },
    /// The source could not be read.
    Io(io::Error),
}

impl fmt::Display for RecordReadError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooLarge { length, limit } => {
                write!(
                    formatter,
                    "record bytes ({length}) exceed the limit ({limit})"
                )
            }
            Self::Io(error) => write!(formatter, "record bytes could not be read: {error}"),
        }
    }
}

impl std::error::Error for RecordReadError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::TooLarge { .. } => None,
            Self::Io(error) => Some(error),
        }
    }
}

impl Record {
    pub(super) fn from_constructed(bytes: Vec<u8>, data_offset: usize) -> Self {
        Self {
            source: RecordSource::Memory(Arc::from(bytes)),
            data_offset,
        }
    }

    /// Own the complete canonical bytes of a record whose header the
    /// validator verified from these same bytes.
    pub(crate) fn from_validated(bytes: Arc<[u8]>, header: &ValidatedHeader) -> Self {
        debug_assert_eq!(bytes.len(), header.canonical_len());
        Self {
            data_offset: header.prefix().len(),
            source: RecordSource::Memory(bytes),
        }
    }

    /// A record over a range of `file` that the validator verified,
    /// `header` describing the bytes at `offset`.
    pub(crate) fn from_validated_file(
        file: Arc<RecordFile>,
        offset: u64,
        header: &ValidatedHeader,
    ) -> Self {
        Self {
            data_offset: header.prefix().len(),
            source: RecordSource::File {
                file,
                offset,
                len: header.canonical_len(),
                prefix: Arc::from(header.prefix()),
            },
        }
    }

    /// Validate complete canonical record bytes (Lace-010) and own them.
    pub fn from_bytes(bytes: impl Into<Vec<u8>>) -> Result<Self, RecordError> {
        let bytes = bytes.into();
        let mut validator = RecordValidator::new();
        if validator.feed(&bytes)? != bytes.len() {
            return Err(RecordError::InvalidFormat(
                "trailing bytes after a complete record".into(),
            ));
        }
        let header = validator.finish()?;
        Ok(Self {
            data_offset: header.prefix().len(),
            source: RecordSource::Memory(Arc::from(bytes)),
        })
    }

    /// # Safety
    ///
    /// The bytes must be immutable canonical bytes admitted after full validation.
    #[doc(hidden)]
    pub unsafe fn try_from_trusted_record_bytes(
        bytes: impl Into<Vec<u8>>,
    ) -> Result<Self, RecordError> {
        let bytes = bytes.into();
        Ok(Self {
            data_offset: trusted_data_offset(&bytes)?,
            source: RecordSource::Memory(Arc::from(bytes)),
        })
    }

    /// A committed record of `len` canonical bytes at `offset` in a store
    /// file. Its prefix is read in at most two bounded reads; hashes and
    /// Marks are not rerun. Malformed trusted bytes are `InvalidData`.
    ///
    /// # Safety
    ///
    /// The range must come from an accepted committed row and hold
    /// immutable canonical bytes admitted after full validation.
    pub(crate) unsafe fn try_from_trusted_file(
        file: Arc<RecordFile>,
        offset: u64,
        len: usize,
    ) -> io::Result<Self> {
        let invalid = |error: RecordError| io::Error::new(io::ErrorKind::InvalidData, error);
        let mut head = vec![0; len.min(FIRST_PREFIX_READ)];
        file.read_exact_at(&mut head, offset)?;
        let data_offset = match trusted_head_data_offset(&head, len).map_err(invalid)? {
            Some(data_offset) => data_offset,
            None => {
                head.resize(len.min(MAX_RECORD_METADATA_BYTES), 0);
                file.read_exact_at(
                    &mut head[FIRST_PREFIX_READ..],
                    offset + FIRST_PREFIX_READ as u64,
                )?;
                trusted_head_data_offset(&head, len)
                    .map_err(invalid)?
                    .ok_or_else(|| invalid(RecordError::InvalidFormat("no record prefix".into())))?
            }
        };
        head.truncate(data_offset);
        Ok(Self {
            data_offset,
            source: RecordSource::File {
                file,
                offset,
                len,
                prefix: Arc::from(head),
            },
        })
    }

    /// # Safety
    ///
    /// The raw mmap range must come from an accepted committed record-log row.
    /// The mapped file must retain its fixed length and identity, and the range
    /// must remain immutable while `mmap` holds its shared arena lock.
    #[cfg(all(target_pointer_width = "64", any(unix, windows)))]
    pub(crate) unsafe fn try_from_trusted_raw_mmap_record_bytes(
        mmap: Arc<ArenaMapping>,
        offset: usize,
        len: usize,
    ) -> Result<Self, RecordError> {
        let end = offset.checked_add(len).ok_or_else(|| {
            RecordError::InvalidFormat("mapped record byte range overflow".into())
        })?;
        if end > mmap.len() {
            return Err(RecordError::InvalidFormat(
                "mapped record byte range exceeds mapping".into(),
            ));
        }
        let source = RecordSource::Mmap { mmap, offset, len };
        Ok(Self {
            data_offset: trusted_data_offset(source.contiguous().expect("mapped bytes"))?,
            source,
        })
    }

    pub fn hash(&self) -> RecordHash {
        self.metadata().hash()
    }

    pub fn kind(&self) -> RecordKind {
        self.hash().kind()
    }

    /// The complete canonical length in bytes.
    pub fn canonical_len(&self) -> usize {
        match &self.source {
            RecordSource::Memory(bytes) => bytes.len(),
            #[cfg(all(target_pointer_width = "64", any(unix, windows)))]
            RecordSource::Mmap { len, .. } => *len,
            RecordSource::File { len, .. } => *len,
        }
    }

    /// The canonical non-payload prefix and everything derived from it.
    pub fn metadata(&self) -> RecordMetadata<'_> {
        let prefix = match &self.source {
            RecordSource::File { prefix, .. } => prefix,
            source => &source.contiguous().expect("contiguous source")[..self.data_offset],
        };
        RecordMetadata::new_trusted(prefix)
    }

    /// The complete canonical bytes, when the source holds them contiguously:
    /// memory and native mmap records, not file ranges.
    pub fn contiguous_bytes(&self) -> Option<&[u8]> {
        self.source.contiguous()
    }

    /// The Blob data, when the source holds it contiguously.
    pub fn contiguous_data(&self) -> Option<&[u8]> {
        Some(&self.source.contiguous()?[self.data_offset..])
    }

    /// Copy canonical bytes starting at `offset` into `buffer` and return how
    /// many were copied: fewer than `buffer.len()` only at the record's end.
    /// A file-range record reads its file at the matching position.
    pub fn read_at(&self, offset: usize, buffer: &mut [u8]) -> io::Result<usize> {
        let len = self.canonical_len();
        if offset > len {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "read offset is beyond the record",
            ));
        }
        let count = (len - offset).min(buffer.len());
        match &self.source {
            RecordSource::File {
                file,
                offset: start,
                ..
            } => file.read_exact_at(&mut buffer[..count], start + offset as u64)?,
            source => {
                let bytes = source.contiguous().expect("contiguous source");
                buffer[..count].copy_from_slice(&bytes[offset..offset + count]);
            }
        }
        Ok(count)
    }

    /// The complete canonical bytes, if at most `limit` bytes long.
    pub fn to_vec(&self, limit: usize) -> Result<Vec<u8>, RecordReadError> {
        self.read_range_to_vec(0, limit)
    }

    /// The Blob data, if at most `limit` bytes long.
    pub fn data_to_vec(&self, limit: usize) -> Result<Vec<u8>, RecordReadError> {
        self.read_range_to_vec(self.data_offset, limit)
    }

    fn read_range_to_vec(&self, start: usize, limit: usize) -> Result<Vec<u8>, RecordReadError> {
        let length = self.canonical_len() - start;
        if length > limit {
            return Err(RecordReadError::TooLarge { length, limit });
        }
        let mut bytes = vec![0; length];
        self.read_at(start, &mut bytes)
            .map_err(RecordReadError::Io)?;
        Ok(bytes)
    }
}