zeppelin-embed 0.8.0

//! Packed private object ownership and explicit abort inventory. Writes alone
//! exclusively creates files, synchronizes artifacts and publishes graph roots.
pub use super::artifact::OwnedArtifact;
use super::artifact::{
    ArtifactIdentity, BlockKind, FramedBlock, MAX_ARTIFACT_BYTES, MAX_BODY_BYTES, PhysicalRef,
    PrivateArtifact,
};
use super::memory::{MAX_STORAGE_PREPARE_BYTES, StorageBuffer, StorageMemory};
use super::tree::directory::{BlockSink, BlockSource, TreeError, TreeResources};
use crate::property_graph::{GraphGeneration, StoreInstanceId};

/// Physical packing limits only; exhausting one pack starts another under the
/// same combined allowance. No logical request is silently split or published.
#[derive(Clone, Copy, Debug)]
pub struct PackLimits {
    /// Complete maximum bytes retained for each object (at most4MiB).
    pub artifact_bytes: usize,
    /// Maximum block descriptors in each packed object.
    pub blocks: usize,
    /// Separate maintenance blocks by their expected lifetime.
    pub streams: PackStreams,
}
impl Default for PackLimits {
    fn default() -> Self {
        Self {
            artifact_bytes: MAX_ARTIFACT_BYTES,
            blocks: (MAX_ARTIFACT_BYTES / 128).min(MAX_BODY_BYTES / 48),
            streams: PackStreams::Single,
        }
    }
}

/// Physical object routing within one private preparation.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum PackStreams {
    /// Commit output shares one stream.
    #[default]
    Single,
    /// Separate stable blocks, graph branches and inventory pages.
    ByLifetime,
}

/// Immutable finalized object bytes that the sole coordinator may protect.
/// Constructed only after every private object has completed validation.
pub struct PreparedArtifact<'a> {
    identity: ArtifactIdentity,
    bytes: &'a [u8],
}
impl<'a> PreparedArtifact<'a> {
    /// Store/object/generation/creation serial retained without reinterpretation.
    pub const fn identity(&self) -> ArtifactIdentity {
        self.identity
    }
    /// Complete finalized file image, including framing and checksum trailer.
    pub const fn bytes(&self) -> &'a [u8] {
        self.bytes
    }
}

/// One complete private preparation's packed objects plus an immutable base.
/// All backing and inventory descriptors retain the same32MiB nested allowance.
/// The identity source is supplied by writes after allocating its checked serial;
/// exclusive filesystem creation remains mandatory before any commit protection.
pub struct PreparedObjects<'a, 'b, S, F> {
    packs: StorageBuffer<'a, PrivateArtifact<'a>>,
    open: [Option<usize>; 3],
    base: &'b S,
    identity_source: F,
    store: StoreInstanceId,
    generation: GraphGeneration,
    limits: PackLimits,
    memory: &'a StorageMemory<'a>,
    finished: bool,
    failed: bool,
}
impl<'a, 'b, S: BlockSource, F: FnMut() -> Result<ArtifactIdentity, TreeError>>
    PreparedObjects<'a, 'b, S, F>
{
    /// Reserves the complete bounded pack/inventory descriptor capacity. No object
    /// identity is requested until the first append needs a new physical pack.
    /// The callback supplies an identity only: a serial may be burned before pack
    /// capacity/allocation succeeds, without transferring a file cleanup duty.
    /// Writes retains ownership of any path it creates before/inside that callback.
    /// Captured callback/base backing remains charged by its external writes owner;
    /// this participant charges its own pack, directory and inventory backing.
    pub fn new(
        base: &'b S,
        identity_source: F,
        store: StoreInstanceId,
        generation: GraphGeneration,
        limits: PackLimits,
        memory: &'a StorageMemory<'a>,
        r: &mut TreeResources<'_>,
    ) -> Result<Self, TreeError> {
        r.require_preparation(memory)?;
        r.step(0)?;
        if !(152..=MAX_ARTIFACT_BYTES).contains(&limits.artifact_bytes)
            || limits.blocks == 0
            || limits.blocks > super::artifact::MAX_BODY_BYTES / 48
        {
            return Err(TreeError::Memory);
        }
        let packs = StorageBuffer::new(memory, MAX_STORAGE_PREPARE_BYTES / limits.artifact_bytes)?;
        Ok(Self {
            packs,
            open: [None; 3],
            base,
            identity_source,
            store,
            generation,
            limits,
            memory,
            finished: false,
            failed: false,
        })
    }
    /// Number of real private object identities currently retained.
    pub fn len(&self) -> usize {
        self.packs.as_slice().len()
    }
    pub(crate) const fn store(&self) -> StoreInstanceId {
        self.store
    }
    pub(crate) const fn generation(&self) -> GraphGeneration {
        self.generation
    }
    pub(crate) const fn memory(&self) -> &'a StorageMemory<'a> {
        self.memory
    }
    /// Whether this preparation allocated no object identity.
    pub fn is_empty(&self) -> bool {
        self.packs.as_slice().is_empty()
    }
    /// Whether every retained pack has completed immutable framing validation.
    pub(crate) const fn is_finished(&self) -> bool {
        self.finished && !self.failed
    }
    /// Exact owned candidates including an interrupted or failed current pack.
    /// This iterator allocates nothing and never includes base/published objects
    /// or identity-only callback results for which no private pack was allocated.
    pub fn abort_inventory(&self) -> impl Iterator<Item = ArtifactIdentity> + '_ {
        self.packs.as_slice().iter().map(PrivateArtifact::identity)
    }
    /// Seal all objects before allowing any complete prepared artifact to escape.
    pub fn finish(&mut self, r: &mut TreeResources<'_>) -> Result<(), TreeError> {
        r.require_preparation(self.memory)?;
        r.step(0)?;
        if self.failed || self.finished {
            return Err(TreeError::Invalid("prepared objects cannot finish"));
        }
        self.failed = true;
        for pack in self.packs.as_mut_slice() {
            if pack.sealed_bytes().is_none() {
                pack.seal(r)?;
            }
        }
        r.step(0)?;
        self.failed = false;
        self.finished = true;
        Ok(())
    }
    /// Borrow one complete prepared artifact only after successful finalization.
    pub fn artifact(&self, index: usize) -> Result<PreparedArtifact<'_>, TreeError> {
        if self.failed || !self.finished {
            return Err(TreeError::Invalid("unfinalized prepared objects"));
        }
        let pack = self
            .packs
            .as_slice()
            .get(index)
            .ok_or(TreeError::Invalid("prepared object index"))?;
        let bytes = pack
            .sealed_bytes()
            .ok_or(TreeError::Invalid("unsealed prepared object"))?;
        Ok(PreparedArtifact {
            identity: pack.identity(),
            bytes,
        })
    }
    fn append_inner(
        &mut self,
        kind: BlockKind,
        bytes: &[u8],
        r: &mut TreeResources<'_>,
    ) -> Result<PhysicalRef, TreeError> {
        if bytes
            .len()
            .checked_add(152)
            .is_none_or(|n| n > self.limits.artifact_bytes)
        {
            return Err(TreeError::Memory);
        }
        let stream = if self.limits.streams == PackStreams::Single || kind != BlockKind::TreePage {
            0
        } else {
            let field = |offset| -> Result<u16, TreeError> {
                let bytes = bytes
                    .get(offset..offset + 2)
                    .ok_or(TreeError::Invalid("short tree page"))?;
                Ok(u16::from_le_bytes(
                    bytes
                        .try_into()
                        .map_err(|_| TreeError::Invalid("tree page field"))?,
                ))
            };
            if field(6)? == super::tree::TreeKind::ObjectInventory as u16 {
                2
            } else if field(16)? > 0 {
                1
            } else {
                0
            }
        };
        let open = *self.open.get(stream).ok_or(TreeError::Memory)?;
        if open
            .and_then(|index| self.packs.as_slice().get(index))
            .is_none_or(|pack| !pack.can_append(bytes.len()))
        {
            if let Some(index) = open {
                self.packs
                    .as_mut_slice()
                    .get_mut(index)
                    .ok_or(TreeError::Memory)?
                    .seal(r)?;
            }
            if self.packs.as_slice().len() == self.packs.capacity() {
                return Err(TreeError::Memory);
            }
            r.step(0)?;
            let identity = (self.identity_source)()?;
            if identity.store != self.store || identity.generation != self.generation {
                return Err(TreeError::Invalid("candidate identity store/generation"));
            }
            if self
                .packs
                .as_slice()
                .iter()
                .map(|pack| pack.identity().creation_serial)
                .max()
                .is_some_and(|serial| serial >= identity.creation_serial)
            {
                return Err(TreeError::Invalid(
                    "candidate creation serial did not advance",
                ));
            }
            for pack in self.packs.as_slice() {
                r.step(1)?;
                if pack.identity().artifact == identity.artifact {
                    return Err(TreeError::Invalid("reused private artifact identity"));
                }
            }
            let pack = PrivateArtifact::new(
                identity,
                self.limits.artifact_bytes,
                self.limits.blocks,
                self.memory,
                r,
            )?;
            let index = self.packs.as_slice().len();
            self.packs.push(pack)?;
            *self.open.get_mut(stream).ok_or(TreeError::Memory)? = Some(index);
        }
        let index = self
            .open
            .get(stream)
            .copied()
            .flatten()
            .ok_or(TreeError::Memory)?;
        self.packs
            .as_mut_slice()
            .get_mut(index)
            .ok_or(TreeError::Memory)?
            .append(kind, bytes, r)
    }
}
impl<S: BlockSource, F> BlockSource for PreparedObjects<'_, '_, S, F> {
    fn page_validated(&self, reference: PhysicalRef) -> Result<bool, TreeError> {
        if self.base.scoped_blocks() {
            return Ok(false);
        }
        if let Some(pack) = self
            .packs
            .as_slice()
            .iter()
            .find(|pack| pack.identity().artifact == reference.artifact)
        {
            return Ok(pack.page_validated(reference));
        }
        self.base.page_validated(reference)
    }

    fn mark_page_validated(&self, reference: PhysicalRef) -> Result<(), TreeError> {
        if self.base.scoped_blocks() {
            return Ok(());
        }
        if let Some(pack) = self
            .packs
            .as_slice()
            .iter()
            .find(|pack| pack.identity().artifact == reference.artifact)
        {
            return pack.mark_page_validated(reference);
        }
        self.base.mark_page_validated(reference)
    }

    fn resolve<'a>(
        &'a self,
        reference: PhysicalRef,
        r: &mut TreeResources<'_>,
    ) -> Result<FramedBlock<'a>, TreeError> {
        r.require_preparation(self.memory)?;
        for pack in self.packs.as_slice() {
            r.step(1)?;
            if pack.identity().artifact == reference.artifact {
                return pack.framed_block(reference, r);
            }
        }
        self.base.resolve(reference, r)
    }
    /// A private pack is already resident, so it is served in place. Every other
    /// reference follows the base source's own retention policy.
    fn with_block<R>(
        &self,
        reference: PhysicalRef,
        r: &mut TreeResources<'_>,
        callback: impl for<'x, 'y> FnOnce(
            FramedBlock<'x>,
            &'y mut TreeResources<'_>,
        ) -> Result<R, TreeError>,
    ) -> Result<R, TreeError> {
        r.require_preparation(self.memory)?;
        for pack in self.packs.as_slice() {
            r.step(1)?;
            if pack.identity().artifact == reference.artifact {
                let block = pack.framed_block(reference, r)?;
                return callback(block, r);
            }
        }
        self.base.with_block(reference, r, callback)
    }
    fn scoped_blocks(&self) -> bool {
        self.base.scoped_blocks()
    }
    fn with_scoped_reads<R>(&self, body: impl FnOnce() -> R) -> R {
        self.base.with_scoped_reads(body)
    }
}
impl<S: BlockSource, F: FnMut() -> Result<ArtifactIdentity, TreeError>> BlockSink
    for PreparedObjects<'_, '_, S, F>
{
    fn append(
        &mut self,
        kind: BlockKind,
        generation: GraphGeneration,
        bytes: &[u8],
        r: &mut TreeResources<'_>,
    ) -> Result<PhysicalRef, TreeError> {
        r.require_preparation(self.memory)?;
        r.step(0)?;
        if self.failed || self.finished || generation != self.generation {
            return Err(TreeError::Invalid(
                "invalid private append state/generation",
            ));
        }
        self.failed = true;
        let reference = self.append_inner(kind, bytes, r)?;
        r.step(0)?;
        self.failed = false;
        Ok(reference)
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::indexing_slicing)]
mod tests {
    use super::super::artifact::{self, ArtifactId, ContainerKind};
    use super::super::tree::{self, Cell, PAGE_BYTES, PageHeader, TreeKind};
    use super::*;
    use crate::lifecycle::{CancelToken, OpenOptions, QueryControl, Store};
    use crate::property_graph::resources::GraphResources;
    use crate::property_graph::staging::{WriteLimits, WriteMemory};

    struct Empty;
    impl BlockSource for Empty {
        fn resolve<'a>(
            &'a self,
            _: PhysicalRef,
            _: &mut TreeResources<'_>,
        ) -> Result<FramedBlock<'a>, TreeError> {
            Err(TreeError::Invalid("empty source"))
        }
    }

    #[test]
    fn private_page_memo_survives_append_and_dies_with_attempt() {
        use super::super::tree::directory::{DirectoryRoot, lookup_entry};
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(
            dir.path(),
            OpenOptions::new().with_max_resident_bytes(256 * 1024 * 1024),
        )
        .unwrap();
        let shared = GraphResources::from_store(&store).unwrap();
        let writer = WriteMemory::new(&shared, WriteLimits::default()).unwrap();
        let control = QueryControl::Cancel(CancelToken::new());
        let memory = StorageMemory::new(&writer, &control, 32 * 1024 * 1024).unwrap();
        let mut r = TreeResources::for_prepare(&memory, u64::MAX).unwrap();
        let id = StoreInstanceId::new(1).unwrap();
        let generation = GraphGeneration::new(1);
        let mut page = [0; PAGE_BYTES];
        tree::encode_page(
            PageHeader {
                kind: TreeKind::Nodes,
                level: 0,
                generation,
            },
            &[],
            &mut page,
        )
        .unwrap();
        for serial in 1..=2 {
            let mut objects = PreparedObjects::new(
                &Empty,
                || {
                    Ok(ArtifactIdentity {
                        store: id,
                        artifact: ArtifactId::new(serial as u128).unwrap(),
                        generation,
                        creation_serial: serial,
                    })
                },
                id,
                generation,
                PackLimits::default(),
                &memory,
                &mut r,
            )
            .unwrap();
            let reference = objects
                .append(BlockKind::TreePage, generation, &page, &mut r)
                .unwrap();
            let root =
                DirectoryRoot::from_reference(id, TreeKind::Nodes, generation, Some(reference))
                    .unwrap();
            let before = shared.work_ledger().unwrap().storage_pages_decoded;
            assert!(
                lookup_entry(&objects, root, &1u128.to_le_bytes(), &mut r)
                    .unwrap()
                    .is_none()
            );
            drop(r);
            assert_eq!(
                shared.work_ledger().unwrap().storage_pages_decoded - before,
                1
            );
            r = TreeResources::for_prepare(&memory, u64::MAX).unwrap();
            // Force descriptor growth; append and sealing must preserve the proof
            // and the exact bytes of the previously admitted private page.
            for _ in 0..512 {
                objects
                    .append(BlockKind::NodeRecord, generation, b"record", &mut r)
                    .unwrap();
            }
            objects.finish(&mut r).unwrap();
            assert_eq!(objects.resolve(reference, &mut r).unwrap().payload(), page);
            let before = shared.work_ledger().unwrap().storage_pages_decoded;
            assert!(
                lookup_entry(&objects, root, &1u128.to_le_bytes(), &mut r)
                    .unwrap()
                    .is_none()
            );
            drop(r);
            assert_eq!(
                shared.work_ledger().unwrap().storage_pages_decoded - before,
                0
            );
            r = TreeResources::for_prepare(&memory, u64::MAX).unwrap();
            assert!(
                !objects
                    .page_validated(PhysicalRef {
                        length: reference.length - 1,
                        ..reference
                    })
                    .unwrap()
            );
        }
    }

    #[test]
    fn by_lifetime_streams_never_mix_block_classes() {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(
            dir.path().join("store"),
            OpenOptions::new().with_max_resident_bytes(256 * 1024 * 1024),
        )
        .unwrap();
        let shared = GraphResources::from_store(&store).unwrap();
        let writer = WriteMemory::new(&shared, WriteLimits::default()).unwrap();
        let control = QueryControl::Cancel(CancelToken::new());
        let memory = StorageMemory::new(&writer, &control, 32 * 1024 * 1024).unwrap();
        let mut r = TreeResources::for_prepare(&memory, u64::MAX).unwrap();
        let id = StoreInstanceId::new(1).unwrap();
        let generation = GraphGeneration::new(1);
        for streams in [PackStreams::Single, PackStreams::ByLifetime] {
            let mut serial = 0;
            let mut objects = PreparedObjects::new(
                &Empty,
                || {
                    serial += 1;
                    Ok(ArtifactIdentity {
                        store: id,
                        artifact: ArtifactId::new(serial as u128).unwrap(),
                        generation,
                        creation_serial: serial,
                    })
                },
                id,
                generation,
                PackLimits {
                    streams,
                    ..PackLimits::default()
                },
                &memory,
                &mut r,
            )
            .unwrap();
            let record = objects
                .append(BlockKind::NodeRecord, generation, b"record", &mut r)
                .unwrap();
            let mut page = [0; PAGE_BYTES];
            tree::encode_page(
                PageHeader {
                    kind: TreeKind::Nodes,
                    level: 0,
                    generation,
                },
                &[],
                &mut page,
            )
            .unwrap();
            let leaf = objects
                .append(BlockKind::TreePage, generation, &page, &mut r)
                .unwrap();
            tree::encode_page(
                PageHeader {
                    kind: TreeKind::Nodes,
                    level: 1,
                    generation,
                },
                &[Cell::Branch {
                    upper: None,
                    child: leaf,
                }],
                &mut page,
            )
            .unwrap();
            let branch = objects
                .append(BlockKind::TreePage, generation, &page, &mut r)
                .unwrap();
            tree::encode_page(
                PageHeader {
                    kind: TreeKind::ObjectInventory,
                    level: 0,
                    generation,
                },
                &[],
                &mut page,
            )
            .unwrap();
            let inventory = objects
                .append(BlockKind::TreePage, generation, &page, &mut r)
                .unwrap();
            // Return to the stable stream after opening both volatile streams.
            let again = objects
                .append(BlockKind::NodeRecord, generation, b"again", &mut r)
                .unwrap();
            objects.finish(&mut r).unwrap();
            eprintln!("{streams:?}: sealed_objects={}", objects.len());
            assert_eq!(
                objects.len(),
                if streams == PackStreams::Single { 1 } else { 3 }
            );
            assert_eq!(record.artifact, leaf.artifact);
            assert_eq!(record.artifact, again.artifact);
            for i in 0..objects.len() {
                let object = objects.artifact(i).unwrap();
                let frame = artifact::decode(ContainerKind::Object, None, object.bytes()).unwrap();
                let classes: std::collections::BTreeSet<_> =
                    [record, leaf, branch, inventory, again]
                        .into_iter()
                        .filter(|reference| reference.artifact == object.identity().artifact)
                        .map(|reference| {
                            let block = frame.framed_block(reference).unwrap();
                            if reference.kind != BlockKind::TreePage {
                                return 0;
                            }
                            let payload = block.payload();
                            if u16::from_le_bytes(payload[6..8].try_into().unwrap())
                                == TreeKind::ObjectInventory as u16
                            {
                                2
                            } else if u16::from_le_bytes(payload[16..18].try_into().unwrap()) > 0 {
                                1
                            } else {
                                0
                            }
                        })
                        .collect();
                if streams == PackStreams::ByLifetime {
                    assert_eq!(classes.len(), 1);
                }
            }
        }
    }
}