riffdb-types 0.1.0

//! One conservative ownership budget for all five exact-provider families.
// req: OQ-040
#![forbid(unsafe_code)]

use std::ops::Deref;
use std::sync::{Arc, Mutex};

pub const EXACT_PROVIDER_BUDGET_BYTES: u64 = 1024 * 1024 * 1024;

#[path = "projection_source_budget.rs"]
mod source_work;
pub use source_work::*;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct BudgetUnavailable;

/// Contains only bounded counters. Identities and paths never enter accounting.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct BudgetObservation {
    pub fixed_owner_bytes: u64,
    pub reserved_bytes: u64,
    pub owned_bytes: u64,
    pub reservations: u64,
    pub allocations: u64,
    pub released_allocations: u64,
    pub active_owners: u64,
    pub retiring_owners: u64,
    pub pinned_owners: u64,
    pub pins: u64,
    pub released_owners: u64,
}

#[path = "exact_provider_lifecycle.rs"]
mod lifecycle;
#[doc(hidden)]
pub use lifecycle::ProviderOwnerAccounting;

#[path = "exact_provider_physical_ownership.rs"]
mod physical;
#[doc(hidden)]
pub use physical::{
    PhysicalOwnershipBalance, PhysicalOwnershipFamily, PhysicalOwnershipFamilyGuard,
    PhysicalOwnershipFamilyObservation, PhysicalOwnershipMember, PreparedPhysicalFamilyAdmission,
    PreparedPhysicalOwnershipIncrement, PreparedSharedPhysicalOwnership,
    SharedPhysicalOwnershipCell,
};

struct Accounting {
    limit: u64,
    counters: BudgetObservation,
    refused: bool,
    source_worker_claimed: bool,
    // Historical aggregate admission, including the fixed accountant owner.
    // Kept outside BudgetObservation: releasing all work restores that baseline.
    peak_admitted_bytes: u64,
}

// Physical control ownership follows this exact accountant Arc, including
// reservations, settled allocations and shared identity charges that escape
// their original engine. The cell contains no data or engine backreference.
struct BudgetOwner {
    accounting: Mutex<Accounting>,
    provider_control: SharedPhysicalOwnershipCell,
}

impl Deref for BudgetOwner {
    type Target = Mutex<Accounting>;
    fn deref(&self) -> &Self::Target {
        &self.accounting
    }
}

const _: () = assert!(
    2 * (std::mem::size_of::<BudgetOwner>() + 2 * std::mem::size_of::<usize>())
        + 2 * std::mem::align_of::<BudgetOwner>()
        + 128
        <= 512
);

#[derive(Clone)]
pub struct ProviderBudget(SharedBudgetOwner);

// Every accountant alias uses this closed owner. Its final physical Arc shell
// dies before the optional foreign provider-control charge inside BudgetOwner.
#[derive(Clone)]
struct SharedBudgetOwner(Option<Arc<BudgetOwner>>);
impl SharedBudgetOwner {
    fn same_owner(&self, other: &Self) -> bool {
        Arc::ptr_eq(
            self.0.as_ref().expect("live budget owner"),
            other.0.as_ref().expect("live budget owner"),
        )
    }
}
impl Deref for SharedBudgetOwner {
    type Target = BudgetOwner;
    fn deref(&self) -> &Self::Target {
        self.0.as_deref().expect("live budget owner")
    }
}
impl Drop for SharedBudgetOwner {
    fn drop(&mut self) {
        if let Some(owner) = self.0.take() {
            drop(Arc::into_inner(owner));
        }
    }
}

impl Default for ProviderBudget {
    fn default() -> Self {
        Self::new()
    }
}

impl ProviderBudget {
    pub fn new() -> Self {
        Self::with_limit_and_owner(EXACT_PROVIDER_BUDGET_BYTES, 512)
    }

    #[cfg(test)]
    fn with_limit(limit: u64) -> Self {
        Self::with_limit_and_owner(limit, 0)
    }

    fn with_limit_and_owner(limit: u64, fixed_owner_bytes: u64) -> Self {
        Self(SharedBudgetOwner(Some(Arc::new(BudgetOwner {
            accounting: Mutex::new(Accounting {
                limit,
                counters: BudgetObservation {
                    fixed_owner_bytes,
                    ..BudgetObservation::default()
                },
                refused: false,
                source_worker_claimed: false,
                peak_admitted_bytes: fixed_owner_bytes,
            }),
            provider_control: SharedPhysicalOwnershipCell::new(),
        }))))
    }

    /// Caller supplies a conservative maximum including allocation capacities
    /// and ownership overhead, before allocating provider/checkpoint/scratch data.
    pub fn reserve(&self, bytes: u64) -> Result<Reservation, BudgetUnavailable> {
        let mut accounting = self.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused || bytes == 0 {
            return Err(BudgetUnavailable);
        }
        let counters = accounting.counters;
        let reserved = counters
            .reserved_bytes
            .checked_add(bytes)
            .ok_or(BudgetUnavailable)?;
        let total = reserved
            .checked_add(counters.owned_bytes)
            .and_then(|bytes| bytes.checked_add(counters.fixed_owner_bytes))
            .ok_or(BudgetUnavailable)?;
        let reservations = counters
            .reservations
            .checked_add(1)
            .ok_or(BudgetUnavailable)?;
        if total > accounting.limit {
            return Err(BudgetUnavailable);
        }
        accounting.counters.reserved_bytes = reserved;
        accounting.counters.reservations = reservations;
        // Every other accountant transition only transfers or reduces this
        // aggregate. Failed admission never changes this historical bound.
        accounting.peak_admitted_bytes = accounting.peak_admitted_bytes.max(total);
        Ok(Reservation {
            budget: self.clone(),
            bytes,
            live: true,
        })
    }

    pub fn observation(&self) -> Result<BudgetObservation, BudgetUnavailable> {
        let accounting = self.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused {
            return Err(BudgetUnavailable);
        }
        Ok(accounting.counters)
    }
}

/// Cancellation releases this caller's unpublished work only.
pub struct Reservation {
    budget: ProviderBudget,
    bytes: u64,
    live: bool,
}

impl Reservation {
    /// Borrow the existing full census allowance; no capacity is admitted here.
    pub fn inspection_scope(
        &mut self,
    ) -> Result<crate::OwnedCapacityInspectionScopeV1<'_>, BudgetUnavailable> {
        self.validate_inspection_scope()
            .map_err(|_| BudgetUnavailable)?;
        Ok(crate::OwnedCapacityInspectionScopeV1(self))
    }

    pub(crate) fn validate_inspection_scope(
        &self,
    ) -> Result<(), crate::OwnedCapacityInspectionErrorV1> {
        if !self.live || self.bytes < crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 {
            return Err(crate::OwnedCapacityInspectionErrorV1);
        }
        let accounting = self
            .budget
            .0
            .lock()
            .map_err(|_| crate::OwnedCapacityInspectionErrorV1)?;
        if accounting.refused {
            return Err(crate::OwnedCapacityInspectionErrorV1);
        }
        Ok(())
    }
    /// Divide already admitted capacity without admitting or allocating new data.
    /// The caller retains a nonzero parent allowance for its root allocation.
    pub fn split(&mut self, bytes: u64) -> Result<Self, BudgetUnavailable> {
        if !self.live || bytes == 0 || bytes >= self.bytes {
            return Err(BudgetUnavailable);
        }
        let mut accounting = self.budget.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused {
            return Err(BudgetUnavailable);
        }
        let reservations = accounting
            .counters
            .reservations
            .checked_add(1)
            .ok_or(BudgetUnavailable)?;
        accounting.counters.reservations = reservations;
        self.bytes -= bytes;
        Ok(Self {
            budget: self.budget.clone(),
            bytes,
            live: true,
        })
    }

    /// Combine unpublished capacity from the same graph without changing the
    /// admitted byte total. Callers combine before constructing the corresponding
    /// page row, so failed transfer cannot release live row data. No owned charge
    /// can be converted back to work through this operation.
    pub fn absorb(&mut self, mut other: Self) -> Result<(), BudgetUnavailable> {
        if !self.live || !other.live || !self.budget.0.same_owner(&other.budget.0) {
            return Err(BudgetUnavailable);
        }
        let bytes = self
            .bytes
            .checked_add(other.bytes)
            .ok_or(BudgetUnavailable)?;
        let mut accounting = self.budget.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused {
            return Err(BudgetUnavailable);
        }
        if accounting.counters.reservations < 2 || accounting.counters.reserved_bytes < bytes {
            accounting.refused = true;
            return Err(BudgetUnavailable);
        }
        accounting.counters.reservations -= 1;
        self.bytes = bytes;
        other.live = false;
        Ok(())
    }

    /// Settle an already admitted, wholly owned candidate after construction.
    /// The original maximum must include inspection scratch before allocation.
    /// Shared payloads require graph admission and are refused by this helper.
    pub fn settle_owned_value<T: crate::InspectOwnedCapacityV1>(
        self,
        value: T,
    ) -> Result<ProviderAllocationOwner<T>, BudgetUnavailable> {
        // Field order also preserves admission on every early-return path.
        let mut candidate = ReservedCandidate { value, work: self };
        let mut scratch = candidate
            .work
            .split(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1)?;
        let report = crate::OwnedCapacityInspectionV1::preview(
            &candidate.value,
            scratch.inspection_scope()?,
        )
        .map_err(|_| BudgetUnavailable)?;
        if !report.shared_allocations().is_empty() {
            return Err(BudgetUnavailable);
        }
        let inline = std::mem::size_of::<ProviderAllocation<T>>()
            .checked_mul(2)
            .and_then(|bytes| bytes.checked_add(2 * std::mem::align_of::<ProviderAllocation<T>>()))
            .and_then(|bytes| bytes.checked_add(1024))
            .and_then(|bytes| u64::try_from(bytes).ok())
            .ok_or(BudgetUnavailable)?;
        let bytes = report
            .unshared_bytes()
            .checked_add(inline)
            .ok_or(BudgetUnavailable)?;
        drop(report);
        drop(scratch);
        // A refused adjustment keeps the work alive through candidate Drop.
        candidate.work.retain_in_place(bytes)?;
        candidate.work.commit_value(candidate.value)
    }

    /// Settle trusted construction containing shared catalog/compiled data.
    /// The caller admits its complete construction maximum, including census
    /// scratch, before allocating. Shared roots require their actual destruction
    /// proof and a matching graph budget; a wrapper cannot invent ownership.
    pub fn settle_shared_value<T: crate::InspectOwnedCapacityV1>(
        self,
        value: T,
    ) -> Result<ProviderAllocationOwner<T>, BudgetUnavailable> {
        let mut candidate = ReservedCandidate { value, work: self };
        let scratch = candidate
            .work
            .split(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1)?;
        let report = crate::OwnedCapacityInspectionV1::inspect_owned(&candidate.value, scratch)
            .map_err(|_| BudgetUnavailable)?;
        let inline = ProviderAllocationOwner::<T>::settlement_root_capacity_upper_bound()?;
        let root_bytes = report
            .unshared_bytes()
            .checked_add(inline)
            .ok_or(BudgetUnavailable)?;
        // First compute the complete requirement without allocating a second
        // discovery vector or attaching a partially admitted graph.
        let mut total = root_bytes;
        for allocation in report.shared_allocations() {
            let lifecycle = allocation.lifecycle().ok_or(BudgetUnavailable)?;
            match lifecycle.attached_owner().map_err(|_| BudgetUnavailable)? {
                Some(owner) => validate_shared_owner(
                    &owner,
                    &candidate.work.budget,
                    allocation.identity(),
                    allocation.bytes(),
                )?,
                None => {
                    total = total
                        .checked_add(allocation.bytes())
                        .ok_or(BudgetUnavailable)?
                }
            }
        }
        // Refuse before any attachment if the prior construction allowance did
        // not cover every root. Settlement never admits additional capacity.
        if total > candidate.work.bytes {
            return Err(BudgetUnavailable);
        }
        for allocation in report.shared_allocations() {
            let lifecycle = allocation.lifecycle().ok_or(BudgetUnavailable)?;
            if let Some(owner) = lifecycle.attached_owner().map_err(|_| BudgetUnavailable)? {
                validate_shared_owner(
                    &owner,
                    &candidate.work.budget,
                    allocation.identity(),
                    allocation.bytes(),
                )?;
                continue;
            }
            let proposed = candidate
                .work
                .split(allocation.bytes())?
                .commit_for_allocation(allocation.identity())?;
            let owner = lifecycle.attach(proposed).map_err(|_| BudgetUnavailable)?;
            validate_shared_owner(
                &owner,
                &candidate.work.budget,
                allocation.identity(),
                allocation.bytes(),
            )?;
        }
        drop(report);
        candidate.work.retain_in_place(root_bytes)?;
        candidate.work.commit_value(candidate.value)
    }

    /// The shared allocation keeps its charge after every discovery reference
    /// has gone. Its data drops before the accounting owner.
    pub fn commit_value<T>(
        mut self,
        value: T,
    ) -> Result<ProviderAllocationOwner<T>, BudgetUnavailable> {
        let minimum = std::mem::size_of::<ProviderAllocation<T>>()
            .checked_add(std::mem::size_of::<OwnershipCharge>())
            .and_then(|bytes| bytes.checked_add(4 * std::mem::size_of::<usize>()))
            .and_then(|bytes| bytes.checked_add(2 * std::mem::align_of::<T>()))
            .and_then(|bytes| u64::try_from(bytes).ok())
            .ok_or(BudgetUnavailable)?;
        if self.bytes < minimum {
            return Err(BudgetUnavailable);
        }
        let charge = self.commit_in_place(None)?;
        Ok(ProviderAllocationOwner(Some(Arc::new(
            ProviderAllocation {
                value,
                _charge: charge,
            },
        ))))
    }
    /// After the work has completed, retain its conservative capacity charge.
    /// Growing here is refused: extra allocation requires prior admission.
    pub fn retain(mut self, bytes: u64) -> Result<Self, BudgetUnavailable> {
        self.retain_in_place(bytes)?;
        Ok(self)
    }

    fn retain_in_place(&mut self, bytes: u64) -> Result<(), BudgetUnavailable> {
        if !self.live || bytes == 0 || bytes > self.bytes {
            return Err(BudgetUnavailable);
        }
        let mut accounting = self.budget.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused {
            return Err(BudgetUnavailable);
        }
        let remaining = accounting
            .counters
            .reserved_bytes
            .checked_sub(self.bytes - bytes)
            .ok_or(BudgetUnavailable)?;
        accounting.counters.reserved_bytes = remaining;
        self.bytes = bytes;
        drop(accounting);
        Ok(())
    }

    /// Transfer the charge to shared ownership. Attach it to the allocation's
    /// final owner, after the charged data fields in destruction order.
    pub fn commit(self) -> Result<SharedOwnershipCharge, BudgetUnavailable> {
        self.commit_with_allocation(None)
    }

    /// Attach only this immutable, checked byte charge to one observed allocation.
    pub fn commit_for_allocation(
        self,
        allocation: crate::SharedAllocationIdentityV1,
    ) -> Result<SharedOwnershipCharge, BudgetUnavailable> {
        self.commit_with_allocation(Some(allocation))
    }

    fn commit_with_allocation(
        mut self,
        allocation: Option<crate::SharedAllocationIdentityV1>,
    ) -> Result<SharedOwnershipCharge, BudgetUnavailable> {
        self.commit_in_place(allocation)
    }

    fn commit_in_place(
        &mut self,
        allocation: Option<crate::SharedAllocationIdentityV1>,
    ) -> Result<SharedOwnershipCharge, BudgetUnavailable> {
        if !self.live {
            return Err(BudgetUnavailable);
        }
        let minimum = std::mem::size_of::<OwnershipCharge>()
            .checked_add(2 * std::mem::size_of::<usize>())
            .and_then(|bytes| bytes.checked_add(std::mem::align_of::<OwnershipCharge>()))
            .and_then(|bytes| u64::try_from(bytes).ok())
            .ok_or(BudgetUnavailable)?;
        if self.bytes < minimum {
            return Err(BudgetUnavailable);
        }
        let mut accounting = self.budget.0.lock().map_err(|_| BudgetUnavailable)?;
        if accounting.refused {
            return Err(BudgetUnavailable);
        }
        let counters = accounting.counters;
        let reserved = counters
            .reserved_bytes
            .checked_sub(self.bytes)
            .ok_or(BudgetUnavailable)?;
        let reservations = counters
            .reservations
            .checked_sub(1)
            .ok_or(BudgetUnavailable)?;
        let owned = counters
            .owned_bytes
            .checked_add(self.bytes)
            .ok_or(BudgetUnavailable)?;
        let allocations = counters
            .allocations
            .checked_add(1)
            .ok_or(BudgetUnavailable)?;
        accounting.counters.reserved_bytes = reserved;
        accounting.counters.reservations = reservations;
        accounting.counters.owned_bytes = owned;
        accounting.counters.allocations = allocations;
        self.live = false;
        drop(accounting);
        Ok(SharedOwnershipCharge(Some(Arc::new(OwnershipCharge {
            budget: self.budget.clone(),
            bytes: self.bytes,
            allocation,
        }))))
    }
}

struct ReservedCandidate<T> {
    value: T,
    // Last: refused inspection/settlement destroys data before releasing work.
    work: Reservation,
}

/// Closed shared allocation ownership. No raw Arc or Weak can escape this
/// handle; final physical shell retirement precedes data and charge destruction.
pub struct ProviderAllocationOwner<T>(Option<Arc<ProviderAllocation<T>>>);
impl<T> ProviderAllocationOwner<T> {
    /// Pure capacity of the closed shared value's root/control allocation.
    /// This is numerical only, does not admit memory or construct an owner.
    /// The sole settlement operation uses exactly this same allowance.
    #[doc(hidden)]
    pub fn settlement_root_capacity_upper_bound() -> Result<u64, BudgetUnavailable> {
        std::mem::size_of::<ProviderAllocation<T>>()
            .checked_mul(2)
            .and_then(|bytes| bytes.checked_add(2 * std::mem::align_of::<ProviderAllocation<T>>()))
            .and_then(|bytes| bytes.checked_add(1024))
            .and_then(|bytes| u64::try_from(bytes).ok())
            .ok_or(BudgetUnavailable)
    }
}

impl<T> ProviderAllocationOwner<T> {
    /// Moves the same allocation into another closed read-only owning handle.
    /// The private value and charge stay together; this performs no allocation.
    pub fn into_closed_allocation(mut self) -> crate::OwnedCapacitySharedV1<ProviderAllocation<T>> {
        crate::OwnedCapacitySharedV1::from_allocation_handle(self.0.take())
    }

    /// Keeps the original allocation and charge while exposing semantic T only.
    pub fn into_erased(mut self) -> crate::OwnedCapacityErasedV1
    where
        T: std::any::Any + Send + Sync,
    {
        crate::OwnedCapacityErasedV1::from_allocation(self.0.take())
    }

    fn get(&self) -> &ProviderAllocation<T> {
        self.0.as_deref().expect("live provider allocation")
    }
    /// Read-only accounting membership; grants no data/storage authority.
    pub fn belongs_to(&self, budget: &ProviderBudget) -> bool {
        self.get().belongs_to(budget)
    }
    /// Conservative settled ownership bound, never extra admission.
    pub fn owned_bytes(&self) -> u64 {
        self.get().owned_bytes()
    }
    /// Exact process-local allocation affinity; no raw pointer is exposed.
    pub fn same_allocation(&self, other: &Self) -> bool {
        Arc::ptr_eq(
            self.0.as_ref().expect("live provider allocation"),
            other.0.as_ref().expect("live provider allocation"),
        )
    }
}
impl<T> Clone for ProviderAllocationOwner<T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}
impl<T> Deref for ProviderAllocationOwner<T> {
    type Target = T;
    fn deref(&self) -> &T {
        &self.get().value
    }
}
impl<T> Drop for ProviderAllocationOwner<T> {
    fn drop(&mut self) {
        if let Some(owner) = self.0.take() {
            drop(Arc::into_inner(owner));
        }
    }
}

pub struct ProviderAllocation<T> {
    value: T,
    // Last: data destruction still owns its byte charge.
    _charge: SharedOwnershipCharge,
}

impl<T> ProviderAllocation<T> {
    /// Accounting membership only; conveys no data or storage authority.
    pub fn belongs_to(&self, budget: &ProviderBudget) -> bool {
        self._charge.belongs_to(budget)
    }

    /// Read-only settled ownership bound for admitting a private successor.
    pub fn owned_bytes(&self) -> u64 {
        self._charge.bytes()
    }
}

impl<T> Deref for ProviderAllocation<T> {
    type Target = T;
    fn deref(&self) -> &T {
        &self.value
    }
}

impl Drop for Reservation {
    fn drop(&mut self) {
        if !self.live {
            return;
        }
        let mut accounting = self
            .budget
            .0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let counters = accounting.counters;
        match (
            counters.reserved_bytes.checked_sub(self.bytes),
            counters.reservations.checked_sub(1),
        ) {
            (Some(bytes), Some(count)) => {
                accounting.counters.reserved_bytes = bytes;
                accounting.counters.reservations = count;
            }
            _ => accounting.refused = true,
        }
    }
}

/// Closed numerical charge owner. Every alias retires its final Arc shell
/// before refunding the existing accountant; it exposes no raw/weak ownership.
#[derive(Clone)]
pub struct SharedOwnershipCharge(Option<Arc<OwnershipCharge>>);
impl SharedOwnershipCharge {
    /// Exact process-local numerical owner affinity only.
    pub fn same_allocation(&self, other: &Self) -> bool {
        Arc::ptr_eq(
            self.0.as_ref().expect("live ownership charge"),
            other.0.as_ref().expect("live ownership charge"),
        )
    }
}
impl Deref for SharedOwnershipCharge {
    type Target = OwnershipCharge;
    fn deref(&self) -> &OwnershipCharge {
        self.0.as_deref().expect("live ownership charge")
    }
}
impl Drop for SharedOwnershipCharge {
    fn drop(&mut self) {
        if let Some(owner) = self.0.take() {
            drop(Arc::into_inner(owner));
        }
    }
}

// The closed wrappers add no allocation or inline storage to the previous Arc
// representation. The existing accountant512 and variable-control formulae
// still price their actual payloads; final shells now retire BEFORE refund.
const _: () = assert!(
    std::mem::size_of::<SharedOwnershipCharge>() == std::mem::size_of::<Arc<OwnershipCharge>>()
);
const _: () = assert!(
    std::mem::align_of::<SharedOwnershipCharge>() == std::mem::align_of::<Arc<OwnershipCharge>>()
);
const _: () =
    assert!(std::mem::size_of::<SharedBudgetOwner>() == std::mem::size_of::<Arc<BudgetOwner>>());
const _: () =
    assert!(std::mem::align_of::<SharedBudgetOwner>() == std::mem::align_of::<Arc<BudgetOwner>>());

/// Discovery removal does not release this charge. Arc destruction does.
pub struct OwnershipCharge {
    budget: ProviderBudget,
    bytes: u64,
    allocation: Option<crate::SharedAllocationIdentityV1>,
}

impl OwnershipCharge {
    /// Process-local binding, never a durable or application identity.
    pub fn allocation(&self) -> Option<crate::SharedAllocationIdentityV1> {
        self.allocation
    }

    pub fn belongs_to(&self, budget: &ProviderBudget) -> bool {
        self.budget.0.same_owner(&budget.0)
    }

    pub fn bytes(&self) -> u64 {
        self.bytes
    }
}

fn validate_shared_owner(
    owner: &OwnershipCharge,
    budget: &ProviderBudget,
    identity: crate::SharedAllocationIdentityV1,
    minimum: u64,
) -> Result<(), BudgetUnavailable> {
    if owner.belongs_to(budget) && owner.allocation() == Some(identity) && owner.bytes() >= minimum
    {
        Ok(())
    } else {
        Err(BudgetUnavailable)
    }
}

impl Drop for OwnershipCharge {
    fn drop(&mut self) {
        let mut accounting = self
            .budget
            .0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let counters = accounting.counters;
        match (
            counters.owned_bytes.checked_sub(self.bytes),
            counters.allocations.checked_sub(1),
            counters.released_allocations.checked_add(1),
        ) {
            (Some(bytes), Some(count), Some(released)) => {
                accounting.counters.owned_bytes = bytes;
                accounting.counters.allocations = count;
                accounting.counters.released_allocations = released;
            }
            _ => accounting.refused = true,
        }
    }
}

#[cfg(test)]
#[path = "exact_provider_owner_retirement_tests.rs"]
mod owner_retirement_tests;

#[cfg(test)]
mod tests {
    use super::*;

    // req: OQ-040
    #[test]
    fn absorbed_reservations_keep_capacity_across_transfer_and_final_capture() {
        use std::sync::Barrier;
        let budget = ProviderBudget::with_limit(512);
        let mut page = budget.reserve(256).unwrap();
        let ready = Arc::new(Barrier::new(2));
        let release = Arc::new(Barrier::new(2));
        let worker = {
            let budget = budget.clone();
            let ready = Arc::clone(&ready);
            let release = Arc::clone(&release);
            std::thread::spawn(move || {
                let row = budget.reserve(256).unwrap();
                ready.wait();
                release.wait();
                row
            })
        };
        ready.wait();
        assert_eq!(budget.observation().unwrap().reservations, 2);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 512);
        release.wait();
        page.absorb(worker.join().unwrap()).unwrap();
        assert_eq!(budget.observation().unwrap().reservations, 1);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 512);
        assert!(budget.reserve(1).is_err());
        let data = page.commit_value(vec![7_u8; 64]).unwrap();
        let capture = data.clone();
        drop(data);
        assert_eq!(budget.observation().unwrap().owned_bytes, 512);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
        assert!(budget.reserve(1).is_err());
        drop(capture);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
        assert_eq!(budget.observation().unwrap().released_allocations, 1);
    }

    // req: OQ-040
    #[test]
    fn absorbing_foreign_budget_refuses_without_releasing_parent_capacity() {
        let budget = ProviderBudget::with_limit(512);
        let foreign = ProviderBudget::with_limit(512);
        let mut page = budget.reserve(256).unwrap();
        assert!(page.absorb(foreign.reserve(128).unwrap()).is_err());
        assert_eq!(foreign.observation().unwrap().reserved_bytes, 0);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 256);
        assert_eq!(budget.observation().unwrap().reservations, 1);
        drop(page);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }

    #[test]
    fn successful_settlement_shrinks_and_retains_charge_through_final_data_drop() {
        use std::sync::atomic::{AtomicBool, Ordering};
        struct Payload {
            bytes: Vec<u8>,
            budget: ProviderBudget,
            destroyed: Arc<AtomicBool>,
        }
        impl crate::InspectOwnedCapacityV1 for Payload {
            fn inspect_owned_capacity(
                &self,
                inspection: &mut crate::OwnedCapacityInspectionV1,
            ) -> Result<(), crate::OwnedCapacityInspectionErrorV1> {
                inspection.visit(&self.bytes)
            }
        }
        impl Drop for Payload {
            fn drop(&mut self) {
                let observed = self.budget.observation().unwrap();
                assert!(observed.owned_bytes >= u64::try_from(self.bytes.capacity()).unwrap());
                assert_eq!(observed.reserved_bytes, 0);
                self.destroyed.store(true, Ordering::Release);
            }
        }
        let budget = ProviderBudget::new();
        let destroyed = Arc::new(AtomicBool::new(false));
        let work = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 32768)
            .unwrap();
        let payload = Payload {
            bytes: Vec::with_capacity(7000),
            budget: budget.clone(),
            destroyed: destroyed.clone(),
        };
        let selected = work.settle_owned_value(payload).unwrap();
        let observed = budget.observation().unwrap();
        assert_eq!(
            (
                observed.reserved_bytes,
                observed.reservations,
                observed.allocations
            ),
            (0, 0, 1)
        );
        assert!(observed.owned_bytes < 32768);
        assert_eq!(selected.owned_bytes(), observed.owned_bytes);
        let captured = selected.clone();
        drop(selected);
        assert!(!destroyed.load(Ordering::Acquire));
        assert_eq!(
            budget.observation().unwrap().owned_bytes,
            observed.owned_bytes
        );
        drop(captured);
        assert!(destroyed.load(Ordering::Acquire));
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    struct SharedCatalogData {
        bytes: Vec<u8>,
        lifecycle: crate::OwnedCapacityLifecycleV1,
    }
    crate::inspect_owned_capacity_struct_v1!(SharedCatalogData {
        owned: [bytes],
        inline: [],
        lifecycle: lifecycle
    });
    struct MixedCatalogCapture {
        scratch: Vec<u8>,
        shared: crate::OwnedCapacitySharedV1<SharedCatalogData>,
    }
    crate::inspect_owned_capacity_struct_v1!(MixedCatalogCapture {
        owned: [scratch, shared],
        inline: []
    });

    #[test]
    fn shared_settlement_retains_distinct_root_and_final_shared_data_charges() {
        let budget = ProviderBudget::new();
        let maximum = crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 65536;
        let work = budget.reserve(maximum).unwrap();
        let shared = crate::OwnedCapacitySharedV1::new(SharedCatalogData {
            bytes: Vec::with_capacity(4096),
            lifecycle: Default::default(),
        });
        let first = work
            .settle_shared_value(MixedCatalogCapture {
                scratch: Vec::with_capacity(2048),
                shared: shared.clone(),
            })
            .unwrap();
        let shared_bytes = shared
            .lifecycle
            .reference()
            .attached_owner()
            .unwrap()
            .unwrap()
            .bytes();
        assert_eq!(
            budget.observation().unwrap().owned_bytes,
            first.owned_bytes() + shared_bytes
        );
        let second = budget
            .reserve(maximum)
            .unwrap()
            .settle_shared_value(MixedCatalogCapture {
                scratch: Vec::with_capacity(2048),
                shared: shared.clone(),
            })
            .unwrap();
        assert_eq!(
            budget.observation().unwrap().owned_bytes,
            first.owned_bytes() + second.owned_bytes() + shared_bytes
        );
        let capture = first.clone();
        drop(first);
        drop(second);
        assert_eq!(
            budget.observation().unwrap().owned_bytes,
            capture.owned_bytes() + shared_bytes
        );
        drop(capture);
        assert_eq!(budget.observation().unwrap().owned_bytes, shared_bytes);
        drop(shared);
        let counters = budget.observation().unwrap();
        assert_eq!(
            (
                counters.owned_bytes,
                counters.reserved_bytes,
                counters.allocations,
                counters.reservations
            ),
            (0, 0, 0, 0)
        );
    }

    #[test]
    fn shared_settlement_checks_complete_allowance_before_attaching_and_rejects_foreign_owners() {
        let budget = ProviderBudget::new();
        let shared = crate::OwnedCapacitySharedV1::new(SharedCatalogData {
            bytes: Vec::with_capacity(4096),
            lifecycle: Default::default(),
        });
        let small = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 2048)
            .unwrap();
        assert!(small.settle_shared_value(shared.clone()).is_err());
        assert!(
            shared
                .lifecycle
                .reference()
                .attached_owner()
                .unwrap()
                .is_none()
        );
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
        let admitted = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 65536)
            .unwrap()
            .settle_shared_value(shared.clone())
            .unwrap();
        let foreign = ProviderBudget::new();
        assert!(
            foreign
                .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 65536)
                .unwrap()
                .settle_shared_value(shared.clone())
                .is_err()
        );
        assert_eq!(foreign.observation().unwrap().owned_bytes, 0);
        assert_eq!(foreign.observation().unwrap().reserved_bytes, 0);
        drop(admitted);
        drop(shared);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn shared_settlement_refuses_missing_destruction_proof() {
        let budget = ProviderBudget::new();
        let maximum = crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 65536;
        assert!(
            budget
                .reserve(maximum)
                .unwrap()
                .settle_shared_value(Arc::new(String::with_capacity(4096)))
                .is_err()
        );
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }

    #[test]
    fn refused_shared_settlement_destroys_candidate_before_releasing_construction_work() {
        use std::sync::atomic::{AtomicBool, Ordering};
        struct ObservedCapture {
            shared: crate::OwnedCapacitySharedV1<SharedCatalogData>,
            budget: ProviderBudget,
            destroyed: Arc<AtomicBool>,
        }
        impl crate::InspectOwnedCapacityV1 for ObservedCapture {
            fn inspect_owned_capacity(
                &self,
                inspection: &mut crate::OwnedCapacityInspectionV1,
            ) -> Result<(), crate::OwnedCapacityInspectionErrorV1> {
                let Self {
                    shared,
                    budget: _,
                    destroyed: _,
                } = self;
                inspection.visit(shared)
            }
        }
        impl Drop for ObservedCapture {
            fn drop(&mut self) {
                assert!(self.budget.observation().unwrap().reserved_bytes > 0);
                self.destroyed.store(true, Ordering::Release);
            }
        }
        let budget = ProviderBudget::new();
        let shared = crate::OwnedCapacitySharedV1::new(SharedCatalogData {
            bytes: Vec::with_capacity(4096),
            lifecycle: Default::default(),
        });
        let destroyed = Arc::new(AtomicBool::new(false));
        let work = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 2048)
            .unwrap();
        assert!(
            work.settle_shared_value(ObservedCapture {
                shared: shared.clone(),
                budget: budget.clone(),
                destroyed: destroyed.clone()
            })
            .is_err()
        );
        assert!(destroyed.load(Ordering::Acquire));
        assert!(
            shared
                .lifecycle
                .reference()
                .attached_owner()
                .unwrap()
                .is_none()
        );
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn wholly_owned_settlement_refuses_shared_payloads_without_publishing() {
        let budget = ProviderBudget::new();
        let work = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 8192)
            .unwrap();
        assert!(
            work.settle_owned_value(Arc::new(String::with_capacity(1024)))
                .is_err()
        );
        let observed = budget.observation().unwrap();
        assert_eq!(
            (
                observed.reserved_bytes,
                observed.owned_bytes,
                observed.allocations
            ),
            (0, 0, 0)
        );
    }

    #[test]
    fn refused_settlement_destroys_candidate_before_releasing_work_allowance() {
        use std::sync::atomic::{AtomicBool, Ordering};
        struct Payload {
            bytes: Vec<u8>,
            budget: ProviderBudget,
            destroyed: Arc<AtomicBool>,
        }
        impl crate::InspectOwnedCapacityV1 for Payload {
            fn inspect_owned_capacity(
                &self,
                inspection: &mut crate::OwnedCapacityInspectionV1,
            ) -> Result<(), crate::OwnedCapacityInspectionErrorV1> {
                inspection.visit(&self.bytes)
            }
        }
        impl Drop for Payload {
            fn drop(&mut self) {
                assert!(self.budget.observation().unwrap().reserved_bytes >= 8192);
                self.destroyed.store(true, Ordering::Release);
            }
        }
        let budget = ProviderBudget::new();
        let destroyed = Arc::new(AtomicBool::new(false));
        let work = budget
            .reserve(crate::OWNED_CAPACITY_INSPECTION_SCRATCH_BYTES_V1 + 8192)
            .unwrap();
        let payload = Payload {
            bytes: Vec::with_capacity(7000),
            budget: budget.clone(),
            destroyed: destroyed.clone(),
        };
        assert!(work.settle_owned_value(payload).is_err());
        assert!(destroyed.load(Ordering::Acquire));
        let observed = budget.observation().unwrap();
        assert_eq!((observed.reserved_bytes, observed.owned_bytes), (0, 0));
    }

    #[test]
    fn split_preserves_total_admission_and_releases_each_child_independently() {
        let budget = ProviderBudget::with_limit(256);
        let mut root = budget.reserve(256).unwrap();
        let child = root.split(128).unwrap();
        let observed = budget.observation().unwrap();
        assert_eq!((observed.reserved_bytes, observed.reservations), (256, 2));
        assert!(budget.reserve(1).is_err());
        let retained = child.commit().unwrap();
        drop(root);
        let observed = budget.observation().unwrap();
        assert_eq!(
            (
                observed.reserved_bytes,
                observed.owned_bytes,
                observed.reservations
            ),
            (0, 128, 0)
        );
        drop(retained);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn invalid_split_cannot_change_parent_admission() {
        let budget = ProviderBudget::with_limit(128);
        let mut root = budget.reserve(128).unwrap();
        for bytes in [0, 128, 129, u64::MAX] {
            assert!(root.split(bytes).is_err());
            let observed = budget.observation().unwrap();
            assert_eq!((observed.reserved_bytes, observed.reservations), (128, 1));
        }
        drop(root);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }

    #[test]
    fn charge_metadata_cannot_be_allocated_from_an_undersized_reservation() {
        let budget = ProviderBudget::with_limit(128);
        assert!(budget.reserve(1).unwrap().commit().is_err());
        let observation = budget.observation().unwrap();
        assert_eq!(
            (observation.reserved_bytes, observation.owned_bytes),
            (0, 0)
        );
        assert!(budget.reserve(128).unwrap().commit().is_ok());
    }

    #[test]
    fn production_budget_is_fixed_at_one_gibibyte() {
        let budget = ProviderBudget::new();
        let fixed = budget.observation().unwrap().fixed_owner_bytes;
        assert!(
            fixed
                >= u64::try_from(
                    std::mem::size_of::<BudgetOwner>() + 2 * std::mem::size_of::<usize>()
                )
                .unwrap()
        );
        let all = budget.reserve(EXACT_PROVIDER_BUDGET_BYTES - fixed).unwrap();
        assert!(budget.reserve(1).is_err());
        drop(all);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }

    #[test]
    fn a_captured_allocation_keeps_its_charge_through_data_destruction() {
        use std::sync::atomic::{AtomicBool, Ordering};
        struct Payload {
            budget: ProviderBudget,
            destroyed: Arc<AtomicBool>,
        }
        impl Drop for Payload {
            fn drop(&mut self) {
                assert_eq!(self.budget.observation().unwrap().owned_bytes, 128);
                self.destroyed.store(true, Ordering::Release);
            }
        }
        let budget = ProviderBudget::with_limit(128);
        let destroyed = Arc::new(AtomicBool::new(false));
        let reservation = budget.reserve(128).unwrap();
        let published = reservation
            .commit_value(Payload {
                budget: budget.clone(),
                destroyed: Arc::clone(&destroyed),
            })
            .unwrap();
        let captured = published.clone();
        drop(published);
        assert!(!destroyed.load(Ordering::Acquire));
        assert!(budget.reserve(1).is_err());
        drop(captured);
        assert!(destroyed.load(Ordering::Acquire));
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn capacity_settlement_cannot_allocate_unreserved_bytes() {
        let budget = ProviderBudget::with_limit(128);
        let work = budget.reserve(128).unwrap();
        let selected = work.retain(64).unwrap().commit().unwrap();
        assert_eq!(budget.observation().unwrap().owned_bytes, 64);
        let other = budget.reserve(64).unwrap();
        assert!(other.retain(65).is_err());
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
        assert_eq!(budget.observation().unwrap().owned_bytes, 64);
        drop(selected);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn reservation_precedes_allocation_and_final_reference_releases_capacity() {
        let budget = ProviderBudget::with_limit(128);
        let reserved = budget.reserve(96).unwrap();
        assert_eq!(budget.observation().unwrap().reserved_bytes, 96);
        assert!(budget.reserve(33).is_err());
        let owned = reserved.commit().unwrap();
        let captured = owned.clone();
        drop(owned);
        assert_eq!(budget.observation().unwrap().owned_bytes, 96);
        assert!(budget.reserve(33).is_err());
        drop(captured);
        let now = budget.observation().unwrap();
        assert_eq!((now.reserved_bytes, now.owned_bytes), (0, 0));
        assert!(budget.reserve(128).is_ok());
    }

    #[test]
    fn failed_and_cancelled_builds_release_only_their_reservation() {
        let budget = ProviderBudget::with_limit(128);
        let selected = budget.reserve(64).unwrap().commit().unwrap();
        let scratch = budget.reserve(64).unwrap();
        assert!(budget.reserve(1).is_err());
        drop(scratch);
        assert_eq!(budget.observation().unwrap().owned_bytes, 64);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
        drop(selected);
        assert_eq!(budget.observation().unwrap().owned_bytes, 0);
    }

    #[test]
    fn overflow_refuses_without_changing_the_existing_charge() {
        let budget = ProviderBudget::with_limit(u64::MAX);
        let first = budget.reserve(u64::MAX - 1).unwrap();
        assert!(budget.reserve(2).is_err());
        assert_eq!(budget.observation().unwrap().reserved_bytes, u64::MAX - 1);
        drop(first);
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }

    #[test]
    fn concurrent_admission_has_one_winner_and_cancel_restores_all_bytes() {
        use std::sync::{Arc, Barrier};
        let budget = ProviderBudget::with_limit(128);
        let start = Arc::new(Barrier::new(3));
        let captured = Arc::new(Barrier::new(3));
        let release = Arc::new(Barrier::new(3));
        let tasks = (0..2)
            .map(|_| {
                let budget = budget.clone();
                let start = start.clone();
                let captured = captured.clone();
                let release = release.clone();
                std::thread::spawn(move || {
                    start.wait();
                    let result = budget.reserve(96);
                    captured.wait();
                    release.wait();
                    result.is_ok()
                })
            })
            .collect::<Vec<_>>();
        start.wait();
        captured.wait();
        assert_eq!(budget.observation().unwrap().reserved_bytes, 96);
        release.wait();
        assert_eq!(
            tasks
                .into_iter()
                .map(|t| t.join().unwrap())
                .filter(|ok| *ok)
                .count(),
            1
        );
        assert_eq!(budget.observation().unwrap().reserved_bytes, 0);
    }
}

#[cfg(test)]
#[path = "exact_provider_lifecycle_tests.rs"]
mod lifecycle_tests;