zeppelin-embed 0.8.0

//! Charged private packing capability; its partial envelope is never published.
use super::*;
use crate::property_graph::storage::memory::{StorageBuffer, StorageMemory};
use crate::property_graph::storage::tree::directory::{TreeError, TreeResources};
use std::cell::Cell;

#[derive(Clone)]
// Constructed only after complete frame/hash validation of this pack's immutable
// prefix. The private owner never exposes mutable bytes or rewrites those blocks.
struct VerifiedEntry {
    reference: PhysicalRef,
    checksum: u64,
    page_validated: Cell<bool>,
}
#[derive(Clone, Copy, Eq, PartialEq)]
enum State {
    Open,
    Failed,
    Sealed,
}

/// An append-only private object under one bounded preparation. Block bytes never
/// change once appended. Only complete sealing grants access to container bytes;
/// cancellation can leave a failed private body requiring whole-object abort.
/// This owns no filesystem path, sync or publication authority.
pub struct PrivateArtifact<'a> {
    bytes: StorageBuffer<'a, u8>,
    entries: StorageBuffer<'a, VerifiedEntry>,
    artifact_limit: usize,
    block_limit: usize,
    identity: ArtifactIdentity,
    memory: &'a StorageMemory<'a>,
    state: State,
}
impl<'a> PrivateArtifact<'a> {
    /// Retain only the initial header; append grows both private buffers on demand.
    pub fn new(
        identity: ArtifactIdentity,
        capacity: usize,
        blocks: usize,
        memory: &'a StorageMemory<'a>,
        r: &mut TreeResources<'_>,
    ) -> Result<Self, TreeError> {
        r.require_preparation(memory)?;
        if !(HEADER_BYTES + 8..=MAX_ARTIFACT_BYTES).contains(&capacity)
            || blocks > MAX_BODY_BYTES / 48
        {
            return Err(TreeError::Memory);
        }
        let mut bytes = StorageBuffer::new(memory, HEADER_BYTES)?;
        let entries = StorageBuffer::new(memory, 0)?;
        r.step(HEADER_BYTES as u64)?;
        bytes.extend_from_slice(&[0; HEADER_BYTES])?;
        Ok(Self {
            bytes,
            entries,
            artifact_limit: capacity,
            block_limit: blocks,
            identity,
            memory,
            state: State::Open,
        })
    }
    /// Exact private allocation identity for the enclosing abort inventory.
    pub const fn identity(&self) -> ArtifactIdentity {
        self.identity
    }
    /// Full retained object and descriptor capacities, including unused backing.
    pub fn owned_bytes(&self) -> usize {
        self.bytes.owned_bytes() + self.entries.owned_bytes()
    }
    /// Whether this input fits without replacing or growing any live allocation.
    pub fn can_append(&self, payload_bytes: usize) -> bool {
        self.state == State::Open
            && self.entries.as_slice().len() < self.block_limit
            && self
                .bytes
                .as_slice()
                .len()
                .checked_add(24)
                .and_then(|n| n.checked_add(payload_bytes))
                .and_then(|n| n.checked_add((self.entries.as_slice().len() + 1) * 24 + 8))
                .is_some_and(|n| n <= self.artifact_limit)
    }

    fn ensure_capacity(
        &mut self,
        bytes_needed: usize,
        entries_needed: usize,
        r: &mut TreeResources<'_>,
    ) -> Result<(), TreeError> {
        const BYTE_STEP: usize = 512 * 1024;
        const ENTRY_STEP: usize = 512;

        let byte_capacity = if bytes_needed > self.bytes.capacity() {
            let rounded = bytes_needed
                .checked_add(BYTE_STEP - 1)
                .ok_or(TreeError::Memory)?
                / BYTE_STEP
                * BYTE_STEP;
            Some(rounded.min(self.artifact_limit))
        } else {
            None
        };
        let entry_capacity = if entries_needed > self.entries.capacity() {
            let doubled = self
                .entries
                .capacity()
                .max(ENTRY_STEP)
                .checked_mul(2)
                .ok_or(TreeError::Memory)?;
            Some(doubled.max(entries_needed).min(self.block_limit))
        } else {
            None
        };

        // Admit every replacement while the old buffers remain live. A failure
        // leaves the current artifact usable and releases any partial reserve.
        let mut next_bytes = byte_capacity
            .map(|capacity| StorageBuffer::new(self.memory, capacity))
            .transpose()?;
        let mut next_entries = entry_capacity
            .map(|capacity| StorageBuffer::new(self.memory, capacity))
            .transpose()?;
        if let Some(bytes) = next_bytes.as_mut() {
            for chunk in self.bytes.as_slice().chunks(65_536) {
                r.step(chunk.len() as u64)?;
                bytes.extend_from_slice(chunk)?;
            }
        }
        if let Some(entries) = next_entries.as_mut() {
            for entry in self.entries.as_slice() {
                r.step(std::mem::size_of::<VerifiedEntry>() as u64)?;
                entries.push(entry.clone())?;
            }
        }
        if let Some(mut replacement) = next_bytes {
            std::mem::swap(&mut self.bytes, &mut replacement);
        }
        if let Some(mut replacement) = next_entries {
            std::mem::swap(&mut self.entries, &mut replacement);
        }
        Ok(())
    }
    /// Append one immutable framed block. Hard-cap refusal leaves this pack
    /// usable; a control error after append admission permanently prevents sealing.
    pub fn append(
        &mut self,
        kind: BlockKind,
        payload: &[u8],
        r: &mut TreeResources<'_>,
    ) -> Result<PhysicalRef, TreeError> {
        r.require_preparation(self.memory)?;
        r.step(0)?;
        if self.state != State::Open {
            return Err(TreeError::Invalid("private artifact is not open"));
        }
        if kind == BlockKind::CheckpointPayload {
            return Err(TreeError::Invalid("checkpoint requires root envelope"));
        }
        if !self.can_append(payload.len()) {
            return Err(TreeError::Memory);
        }
        let checksum =
            controlled_checksum(payload, &mut |n| r.step(n as u64)).map_err(tree_error)?;
        let body_end = self
            .bytes
            .as_slice()
            .len()
            .checked_add(24)
            .and_then(|n| n.checked_add(payload.len()))
            .ok_or(TreeError::Memory)?;
        let directory_and_trailer = self
            .entries
            .as_slice()
            .len()
            .checked_add(1)
            .and_then(|n| n.checked_mul(24))
            .and_then(|n| body_end.checked_add(n))
            .and_then(|n| n.checked_add(8))
            .ok_or(TreeError::Memory)?;
        if let Err(error) =
            self.ensure_capacity(directory_and_trailer, self.entries.as_slice().len() + 1, r)
        {
            if !matches!(error, TreeError::Memory) {
                self.state = State::Failed;
            }
            return Err(error);
        }
        self.state = State::Failed;
        let reference = PhysicalRef {
            artifact: self.identity.artifact,
            offset: self.bytes.as_slice().len() as u64,
            length: u32::try_from(payload.len() + 24).map_err(|_| TreeError::Memory)?,
            kind,
            version: 1,
        };
        let mut header = [0; 24];
        put(&mut header, 0, &(kind as u16).to_le_bytes())?;
        put(&mut header, 2, &1u16.to_le_bytes())?;
        put(&mut header, 8, &(payload.len() as u64).to_le_bytes())?;
        put(&mut header, 16, &checksum.to_le_bytes())?;
        r.step(24)?;
        self.bytes.extend_from_slice(&header)?;
        for chunk in payload.chunks(65_536) {
            r.step(chunk.len() as u64)?;
            self.bytes.extend_from_slice(chunk)?;
        }
        validate_block(self.bytes.as_slice(), reference, checksum, &mut |n| {
            r.step(n as u64)
        })
        .map_err(tree_error)?;
        r.step(std::mem::size_of::<VerifiedEntry>() as u64)?;
        self.entries.push(VerifiedEntry {
            reference,
            checksum,
            page_validated: Cell::new(false),
        })?;
        r.step(0)?;
        self.state = State::Open;
        Ok(reference)
    }

    // The owning attempt fixes the full artifact identity. Each admitted entry
    // fixes the offset and exact extent; append/growth/seal never rewrite its
    // payload. The proof is charged with the descriptor and dies with the pack.
    fn page_proof(&self, reference: PhysicalRef) -> Option<&Cell<bool>> {
        if self.state == State::Failed
            || reference.artifact != self.identity.artifact
            || reference.kind != BlockKind::TreePage
        {
            return None;
        }
        let index = self
            .entries
            .as_slice()
            .binary_search_by_key(&reference.offset, |entry| entry.reference.offset)
            .ok()?;
        self.entries
            .as_slice()
            .get(index)
            .filter(|entry| entry.reference == reference)
            .map(|entry| &entry.page_validated)
    }

    pub(crate) fn page_validated(&self, reference: PhysicalRef) -> bool {
        self.page_proof(reference).is_some_and(Cell::get)
    }

    pub(crate) fn mark_page_validated(&self, reference: PhysicalRef) -> Result<(), TreeError> {
        let proof = self
            .page_proof(reference)
            .ok_or(TreeError::Invalid("private page proof reference"))?;
        proof.set(true);
        Ok(())
    }
    /// Resolve an exact reference admitted by complete append-time frame/hash
    /// validation. Later appends and sealing preserve its immutable payload bytes.
    /// Rust borrowing forbids an append while these immutable bytes are borrowed.
    pub fn framed_block(
        &self,
        reference: PhysicalRef,
        r: &mut TreeResources<'_>,
    ) -> Result<FramedBlock<'_>, TreeError> {
        r.require_preparation(self.memory)?;
        if self.state == State::Failed || reference.artifact != self.identity.artifact {
            return Err(TreeError::Invalid("failed or substituted private artifact"));
        }
        let (mut low, mut high) = (0, self.entries.as_slice().len());
        while low < high {
            r.step(std::mem::size_of::<VerifiedEntry>() as u64)?;
            let middle = low + (high - low) / 2;
            let entry = self
                .entries
                .as_slice()
                .get(middle)
                .ok_or(TreeError::Memory)?;
            match entry.reference.offset.cmp(&reference.offset) {
                std::cmp::Ordering::Less => low = middle + 1,
                std::cmp::Ordering::Greater => high = middle,
                std::cmp::Ordering::Equal if entry.reference == reference => {
                    // Entry construction is the complete validation boundary;
                    // exact reference matching binds it to this immutable prefix.
                    let start = usize_from(reference.offset)?;
                    let payload = self
                        .bytes
                        .as_slice()
                        .get(add(start, 24)?..add(start, reference.length as usize)?)
                        .ok_or(TreeError::Invalid("admitted private block extent"))?;
                    let trailer = self
                        .bytes
                        .as_slice()
                        .len()
                        .checked_sub(8)
                        .ok_or(TreeError::Invalid("admitted private artifact trailer"))?;
                    let file_checksum =
                        frame::read_u64("native graph artifact", self.bytes.as_slice(), trailer)?;
                    r.step(0)?;
                    return Ok(FramedBlock {
                        identity: self.identity,
                        reference,
                        payload,
                        file_length: self.bytes.as_slice().len(),
                        file_checksum,
                    });
                }
                std::cmp::Ordering::Equal => break,
            }
        }
        Err(TreeError::Invalid(
            "private reference is not an exact directory entry",
        ))
    }
    /// Finish the one immutable container and verify the shared complete codec.
    /// Failure retains this identity and private bytes for explicit abort handling.
    pub fn seal(&mut self, r: &mut TreeResources<'_>) -> Result<(), TreeError> {
        r.require_preparation(self.memory)?;
        r.step(0)?;
        if self.state != State::Open {
            return Err(TreeError::Invalid("private artifact cannot seal"));
        }
        let seal_length = self
            .entries
            .as_slice()
            .len()
            .checked_mul(24)
            .and_then(|n| self.bytes.as_slice().len().checked_add(n))
            .and_then(|n| n.checked_add(8))
            .ok_or(TreeError::Memory)?;
        self.ensure_capacity(seal_length, self.entries.as_slice().len(), r)?;
        self.state = State::Failed;
        let body = self.bytes.as_slice().len() - HEADER_BYTES;
        for entry in self.entries.as_slice() {
            let mut bytes = [0; 24];
            put(&mut bytes, 0, &entry.reference.offset.to_le_bytes())?;
            put(&mut bytes, 8, &entry.reference.length.to_le_bytes())?;
            put(&mut bytes, 12, &(entry.reference.kind as u16).to_le_bytes())?;
            put(&mut bytes, 14, &entry.reference.version.to_le_bytes())?;
            put(&mut bytes, 16, &entry.checksum.to_le_bytes())?;
            r.step(24)?;
            self.bytes.extend_from_slice(&bytes)?;
        }
        let length = self
            .bytes
            .as_slice()
            .len()
            .checked_add(8)
            .ok_or(TreeError::Memory)?;
        r.step(HEADER_BYTES as u64)?;
        encode_header(
            ContainerKind::Object,
            self.identity,
            length,
            body,
            self.entries.as_slice().len(),
            self.bytes.as_mut_slice(),
        )?;
        let checksum = controlled_checksum(self.bytes.as_slice(), &mut |n| r.step(n as u64))
            .map_err(tree_error)?;
        r.step(8)?;
        self.bytes.extend_from_slice(&checksum.to_le_bytes())?;
        decode_with_control(
            ContainerKind::Object,
            Some((self.identity.store, self.identity.artifact)),
            self.bytes.as_slice(),
            &mut |n| r.step(n as u64),
        )
        .map_err(tree_error)?;
        r.step(0)?;
        self.state = State::Sealed;
        Ok(())
    }
    /// Complete immutable bytes only after successful final validation.
    pub fn sealed_bytes(&self) -> Option<&[u8]> {
        (self.state == State::Sealed).then(|| self.bytes.as_slice())
    }
}
fn tree_error(error: ArtifactControlError<TreeError>) -> TreeError {
    match error {
        ArtifactControlError::Format(error) => TreeError::Format(error),
        ArtifactControlError::Control(error) => error,
    }
}

/// Fully framed immutable bytes retained under preparation ownership. This is
/// an artifact validation capability, not a coherent graph read-view lease.
/// Its complete decoder is run exactly once before cached resolution is allowed.
pub struct OwnedArtifact<'a> {
    bytes: StorageBuffer<'a, u8>,
    identity: ArtifactIdentity,
    directory: usize,
    count: usize,
}
impl<'a> OwnedArtifact<'a> {
    /// Read one complete physical object in controlled <=64KiB spans. A length
    /// mismatch, trailing byte, identity mismatch or malformed frame rejects.
    pub fn read_from(
        reader: &mut impl std::io::Read,
        length: usize,
        expected: (StoreInstanceId, ArtifactId),
        memory: &'a StorageMemory<'a>,
        r: &mut TreeResources<'_>,
    ) -> Result<Self, TreeError> {
        r.require_preparation(memory)?;
        if !(HEADER_BYTES + 8..=MAX_ARTIFACT_BYTES).contains(&length) {
            return Err(TreeError::Invalid("owned artifact length"));
        }
        let mut bytes = StorageBuffer::new(memory, length)?;
        let mut scratch = [0; 65_536];
        while bytes.as_slice().len() < length {
            let count = (length - bytes.as_slice().len()).min(scratch.len());
            let chunk = scratch.get_mut(..count).ok_or(TreeError::Memory)?;
            // read_exact retries Interrupted internally without a checkpoint.
            // Poll each physical attempt, including zero-progress interruptions,
            // and charge only the bytes the reader actually returned.
            r.step(0)?;
            let read = match reader.read(chunk) {
                Ok(0) => return Err(TreeError::Io(std::io::ErrorKind::UnexpectedEof.into())),
                Ok(read) => read,
                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(error) => return Err(TreeError::Io(error)),
            };
            let chunk = chunk
                .get(..read)
                .ok_or(TreeError::Invalid("reader exceeded output"))?;
            r.step(read as u64)?;
            r.step(read as u64)?;
            bytes.extend_from_slice(chunk)?;
        }
        let mut trailing = [0];
        loop {
            r.step(0)?;
            match reader.read(&mut trailing) {
                Ok(0) => break,
                Ok(_) => return Err(TreeError::Invalid("trailing artifact bytes")),
                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(error) => return Err(TreeError::Io(error)),
            }
        }
        let frame = decode_with_control(
            ContainerKind::Object,
            Some(expected),
            bytes.as_slice(),
            &mut |n| r.step(n as u64),
        )
        .map_err(tree_error)?;
        let (identity, directory, count) = (frame.identity, frame.directory, frame.count);
        r.step(0)?;
        Ok(Self {
            bytes,
            identity,
            directory,
            count,
        })
    }
    /// Exact identity validated before this owner was constructed.
    pub const fn identity(&self) -> ArtifactIdentity {
        self.identity
    }
    /// Complete actual retained backing capacity.
    pub fn owned_bytes(&self) -> usize {
        self.bytes.owned_bytes()
    }
}
impl crate::property_graph::storage::tree::directory::BlockSource for OwnedArtifact<'_> {
    fn resolve<'a>(
        &'a self,
        reference: PhysicalRef,
        r: &mut TreeResources<'_>,
    ) -> Result<FramedBlock<'a>, TreeError> {
        // One bounded binary search over immutable already validated entries.
        // There is no disk access, allocation or new view admission here.
        r.step(1)?;
        let frame = ArtifactFrame {
            bytes: self.bytes.as_slice(),
            identity: self.identity,
            kind: ContainerKind::Object,
            directory: self.directory,
            count: self.count,
            file_checksum: frame::read_u64(
                "native graph artifact",
                self.bytes.as_slice(),
                self.bytes
                    .as_slice()
                    .len()
                    .checked_sub(8)
                    .ok_or(TreeError::Invalid("owned artifact trailer"))?,
            )?,
        };
        let block = frame.framed_block(reference)?;
        r.step(0)?;
        Ok(block)
    }
}