o3 0.31.1

pub(in crate::buffer::pool) mod lease;
use std::{marker, mem, ptr};

use crate::{
    buffer,
    buffer::{
        pool,
        pool::{owners, state},
        resident::{AllocError, staging, write::Cause},
        storage,
        write::{consume, consume::Error},
    },
};

pub(in crate::buffer::pool) mod append;
pub(in crate::buffer::pool) mod frozen;

/// Owns one live slot edge and its initialized prefix. The backing and prefix
/// cannot be replaced independently through this capability.
#[derive(Debug)]
#[repr(C)]
pub struct Span<K: crate::buffer::raw::Backing = owners::Plain, M = pool::Owned, S = state::Uninit>
{
    pub(in crate::buffer::pool) slots: owners::Ptr<K, S>,
    pub(in crate::buffer::pool) key: owners::Key,
    pub(in crate::buffer::pool) len: u32,
    pub(in crate::buffer::pool) root: M,
}

impl<K: crate::buffer::raw::Backing, S> Span<K, pool::Owned, S> {
    pub fn recycle(self) {
        drop(self);
    }

    pub(in crate::buffer::pool) fn acquire(core: owners::Ptr<K, S>) -> Option<Self> {
        let slots = core;
        let claim = owners::Claim::new(slots)?;
        Some(Self::new(claim.take(), pool::Owned))
    }

    pub(in crate::buffer::pool) fn from_seed(seed: owners::Seed<K, S>) -> Self {
        let (core, active) = seed.take();
        // Seed transferred the sole root; active keeps the pool live through detachment.
        owners::Header::detach(core);
        Self::new(active, pool::Owned)
    }
}

impl<'pool, K: crate::buffer::raw::Backing, S> Span<K, pool::Borrowed<'pool>, S> {
    pub(in crate::buffer::pool) fn borrow(core: &'pool owners::Ptr<K, S>) -> Option<Self> {
        let slots = *core;
        let claim = owners::Claim::new(slots)?;
        Some(Self::new(claim.take(), pool::Borrowed(marker::PhantomData)))
    }
}

impl<K: crate::buffer::raw::Backing, M, S> Span<K, M, S> {
    pub(in crate::buffer) fn into_edge(self) -> owners::Edge<K, M> {
        let this = mem::ManuallyDrop::new(self);
        owners::Edge::new(this.entry().owner())
    }

    pub(in crate::buffer::pool) fn new(active: owners::Active<K, S>, root: M) -> Self {
        let active = mem::ManuallyDrop::new(active);
        Self {
            slots: active.slots,
            key: active.key,
            len: 0,
            root,
        }
    }
}

impl<'d> Span<owners::Resident<'d>> {
    pub(in crate::buffer) fn try_staged<I: buffer::raw::SliceExt>(
        reserve: staging::Reserve<&staging::Pool<'d>, (&[u8], owners::Measured<I>)>,
    ) -> Result<Self, AllocError> {
        let (pool, capacity, (first, second)) = reserve.into_parts();
        let len = first.len() + second.len();
        let layout = pool::NonEmpty::new(capacity);
        let seed =
            owners::Block::<owners::Accounted, state::Uninit>::try_seed(layout, pool.budget())?;
        let span = Span::from_seed(seed);
        Ok(span.copy_in((first, second.into_input()), len))
    }
}

const _: () = {
    assert!(size_of::<Span>() == 2 * size_of::<usize>());
    assert!(size_of::<Span<owners::Plain, pool::Borrowed<'static>>>() == 2 * size_of::<usize>());
    assert!(size_of::<Span<owners::Resident<'static>>>() == 2 * size_of::<usize>());
    assert!(
        size_of::<Span<owners::Plain, pool::Owned, state::Initialized>>() == 2 * size_of::<usize>()
    );
};

impl<K: crate::buffer::raw::Backing, M, S> AsRef<[u8]> for Span<K, M, S> {
    fn as_ref(&self) -> &[u8] {
        self.as_slice()
    }
}

impl<S: state::State, L> pool::Span<owners::Plain<L>, pool::Owned, S> {
    #[must_use]
    pub fn freeze(self) -> pool::Span<owners::Plain<L>, pool::Owned, pool::Frozen> {
        let this = mem::ManuallyDrop::new(self);
        Span {
            slots: this.slots.erase_state(),
            key: this.key,
            len: this.len,
            root: this.root,
        }
    }
}

impl<'pool, S: state::State, L> pool::Span<owners::Plain<L>, pool::Borrowed<'pool>, S> {
    #[must_use]
    pub fn freeze(self) -> storage::Shared<owners::Edge<owners::Plain<L>, pool::Borrowed<'pool>>> {
        let len = self.len();
        // SAFETY: consuming the span retains its exact initialized prefix.
        unsafe { pool::raw::into_shared(self, 0, len) }
    }
}

impl<L, M> pool::Span<owners::Plain<L>, M, state::Uninit> {
    pub fn try_append<'a, 'source>(
        &'a mut self,
        source: &'source [u8],
    ) -> Result<pool::Append<'a, &'source [u8], Self>, Error> {
        let input = owners::Measured::slice(source);
        pool::Append::try_lease(self, input)
    }
}

impl pool::Span<owners::Plain<pool::Layout>, pool::Owned, state::Uninit> {
    /// Allocates one independently owned writable slot.
    pub fn try_new(cap: usize) -> Result<Self, pool::CreateError> {
        let layout = pool::NonEmpty::one(cap).map_err(pool::CreateError::Layout)?;
        let seed = owners::Block::<owners::Plain<pool::Layout>, state::Uninit>::try_seed(layout)
            .map_err(pool::CreateError::Exhausted)?;
        Ok(pool::Span::from_seed(seed))
    }

    pub fn try_copy<const N: usize>(
        mut self,
        source: buffer::raw::Slices<'_, N>,
    ) -> Result<pool::Span<owners::Plain, pool::Owned, pool::Frozen>, Error> {
        let start = self.len();
        let len = source.len().get();
        Error::check(len as usize, self.capacity().get() as usize - start)?;
        let target = self.entry().data();
        // SAFETY: admission retains every source and this exact exclusive suffix.
        unsafe { buffer::raw::SliceExt::copy_to(source.slices(), target.add(start)) };
        self.len += len;
        Ok(self.freeze())
    }
}

impl<L> pool::Span<owners::Plain<L>, pool::Owned, state::Initialized> {
    pub fn output(&mut self) -> buffer::raw::Output<'_, L> {
        buffer::raw::Output::new(self)
    }
}

/// One acquired resident-pool backing without a duplicate pool owner.
impl<'d> pool::Span<owners::Resident<'d>> {
    pub(in crate::buffer) fn try_extend<'a, I: buffer::raw::SliceExt>(
        slot: staging::Slot<'a, 'd>,
        slab: Option<&pool::Pool<state::Uninit, owners::Resident<'d>>>,
        input: &owners::Measured<I>,
    ) -> Result<(), Cause> {
        let added = input.len();
        let src = input.input();
        let (cursor, head) = slot.into_parts();
        let capacity = cursor.capacity().get() as usize;
        let tail = match consume::Error::check(added, capacity - cursor.len()) {
            Ok(()) => {
                let start = cursor.len();
                let target = cursor.entry().data();
                let end = start + added;
                // SAFETY: admission retains the exact suffix and every disjoint source.
                unsafe { src.copy_to(target.add(start)) };
                cursor.len = end as u32;
                return Ok(());
            }
            Err(error) => error,
        };
        let len = cursor.len() - *head as usize;
        consume::Error::check(added, capacity - len).map_err(Cause::Capacity)?;
        if cursor.entry().is_unique() {
            let data = cursor.entry().data();
            let start = mem::take(head) as usize;
            // SAFETY: the exclusive slot is unique and both ranges remain within its initialized extent.
            unsafe { ptr::copy(data.add(start), data, len) };
            let end = len + added;
            // SAFETY: compaction preserves the admitted suffix and disjoint sources.
            unsafe { src.copy_to(data.add(len)) };
            cursor.len = end as u32;
            return Ok(());
        }
        let Some(slab) = slab else {
            return Err(Cause::Shared(tail));
        };
        // Slot and slab retain the same warm pool; the suffix check above proves the copy fits.
        let Some(next) = slab.acquire() else {
            return Err(Cause::Rotation(tail));
        };
        // SAFETY: the exclusive source borrow retains its exact suffix through the copy.
        let bytes = unsafe { cursor.as_slice().get_unchecked(*head as usize..) };
        let next = next.copy_in((bytes, src), bytes.len() + added);
        *head = 0;
        *cursor = next;
        Ok(())
    }
}