uqa-storage 0.4.8

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Monotonic identifier reservations are independent of record visibility and transaction undo.

mod conformance;
pub use conformance::{verify_identifier_allocations, verify_identifier_batches};

use std::{num::NonZeroU64, ops::RangeInclusive};

use uqa_core::memory::{MemoryError, MemoryReservation};

use crate::{read_control::StorageReadControl, StorageBackendResult};

use super::{VersionError, VersionResult, RECLAMATION_DOMAIN_PREFIX, RECLAMATION_EPOCH_NAMESPACE};

/// Session-bound access to durable, nontransactional identifier reservations. Implementations retain their session's cancellation, memory and read-only checks; forwarding storage wrappers must preserve this capability.
pub trait IdentifierAllocator: Send + Sync {
    /// Read the current autonomous watermark without reserving an identity or creating a namespace. This read is independent of the logical record snapshot and remains available to read-only sessions.
    fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>>;

    fn allocate_identifiers(
        &self,
        namespace: &[u8],
        request: IdentifierRequest,
    ) -> StorageBackendResult<IdentifierAllocation>;

    /// Raise the watermark to at least `value` without reporting it. A session that has already read a watermark at or above `value` answers without a physical allocation, which an `Observe` allocation cannot do because it reports the current watermark. Forwarding wrappers forward this as well, or every such observation stays physical.
    fn observe_identifier(
        &self,
        namespace: &[u8],
        value: u64,
    ) -> StorageBackendResult<ObservedIdentifier> {
        self.allocate_identifiers(namespace, IdentifierRequest::Observe(value))
            .map(|allocation| ObservedIdentifier::Observed {
                previous: allocation.previous(),
            })
    }
}

/// What a session learned when it raised a watermark to an identity.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ObservedIdentifier {
    /// A watermark the session had read already covers the identity. Where the watermark stood before is not known.
    Covered,
    /// The watermark was read and raised under the provider's write admission. `previous` is where it stood, `None` for a namespace that had none.
    Observed { previous: Option<u64> },
}

impl ObservedIdentifier {
    /// Whether `identity` was never used in its namespace before this observation. Every use of an identity raises the watermark to it, so an identity above the earlier watermark was not used.
    pub fn unused_before(self, identity: u64) -> bool {
        match self {
            Self::Covered => false,
            Self::Observed { previous } => previous.is_none_or(|previous| identity > previous),
        }
    }
}

/// Observe an externally supplied identity or reserve a contiguous range. Namespace keys must include the allocation domain and the owning object's non-reused generation, rather than its reusable name.
#[derive(Debug, Clone, Copy)]
pub enum IdentifierRequest {
    Observe(u64),
    Reserve {
        minimum: u64,
        maximum: u64,
        count: NonZeroU64,
    },
}

/// The new durable high watermark and, for a reservation, its inclusive allocated range.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IdentifierAllocation {
    watermark: u64,
    first: Option<u64>,
    previous: Option<u64>,
}

impl IdentifierAllocation {
    pub const fn watermark(self) -> u64 {
        self.watermark
    }

    /// The watermark before this allocation, or `None` when the namespace had none.
    pub const fn previous(self) -> Option<u64> {
        self.previous
    }

    pub fn range(self) -> Option<RangeInclusive<u64>> {
        self.first.map(|first| first..=self.watermark)
    }
}

impl IdentifierRequest {
    /// Compute the next state while the provider holds exclusive physical admission. Neither observation nor reservation ever lowers an existing watermark; exhaustion leaves it unchanged.
    pub fn prepare(self, current: Option<u64>) -> VersionResult<IdentifierAllocation> {
        let (watermark, first) = match self {
            Self::Observe(value) => (current.map_or(value, |current| current.max(value)), None),
            Self::Reserve {
                minimum,
                maximum,
                count,
            } => {
                if minimum > maximum {
                    return Err(VersionError::InvalidIdentifierBounds);
                }
                let first = match current {
                    Some(current) => current
                        .checked_add(1)
                        .ok_or(VersionError::IdentifiersExhausted)?
                        .max(minimum),
                    None => minimum,
                };
                let last = first
                    .checked_add(count.get() - 1)
                    .filter(|last| *last <= maximum)
                    .ok_or(VersionError::IdentifiersExhausted)?;
                (last, Some(first))
            }
        };
        Ok(IdentifierAllocation {
            watermark,
            first,
            previous: current,
        })
    }

    /// Validate a request before physical admission and charge the provider's key binding/copy workspace. No identifier is consumed if this fails.
    pub fn reserve_workspace(
        self,
        namespace: &[u8],
        control: &StorageReadControl,
    ) -> VersionResult<MemoryReservation> {
        self.prepare(None)?;
        reserve_identifier_workspace(namespace, control)
    }
}

/// Charge namespace bindings for either a reservation or a read without retaining the caller's bytes.
pub fn reserve_identifier_workspace(
    namespace: &[u8],
    control: &StorageReadControl,
) -> VersionResult<MemoryReservation> {
    control.cancellation().check()?;
    if namespace.is_empty() {
        return Err(VersionError::InvalidEncoding("empty identifier namespace"));
    }
    if namespace == RECLAMATION_EPOCH_NAMESPACE || namespace.starts_with(RECLAMATION_DOMAIN_PREFIX)
    {
        return Err(VersionError::InvalidEncoding(
            "reserved reclamation namespace",
        ));
    }
    let bytes = namespace
        .len()
        .checked_mul(2)
        .and_then(|bytes| bytes.checked_add(64))
        .ok_or(MemoryError::SizeOverflow)?;
    Ok(control.memory().reserve(bytes)?)
}

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

    fn reserve(minimum: u64, maximum: u64, count: u64) -> IdentifierRequest {
        IdentifierRequest::Reserve {
            minimum,
            maximum,
            count: NonZeroU64::new(count).unwrap(),
        }
    }

    #[test]
    fn reservations_cover_zero_full_width_bounds_and_atomic_exhaustion() {
        assert_eq!(reserve(0, 9, 2).prepare(None).unwrap().range(), Some(0..=1));
        assert_eq!(
            reserve(5, 9, 2).prepare(Some(1)).unwrap().range(),
            Some(5..=6)
        );
        assert_eq!(
            reserve(0, u64::MAX, 2)
                .prepare(Some(u64::MAX - 2))
                .unwrap()
                .range(),
            Some(u64::MAX - 1..=u64::MAX)
        );
        for (request, current) in [
            (reserve(0, 9, 2), Some(8)),
            (reserve(0, u64::MAX, 2), Some(u64::MAX - 1)),
            (reserve(0, u64::MAX, 1), Some(u64::MAX)),
        ] {
            assert!(matches!(
                request.prepare(current),
                Err(VersionError::IdentifiersExhausted)
            ));
        }
        assert!(matches!(
            reserve(2, 1, 1).prepare(None),
            Err(VersionError::InvalidIdentifierBounds)
        ));
    }

    #[test]
    fn observations_never_rewind_or_wrap_a_namespace() {
        for (current, observed, expected) in [
            (None, 0, 0),
            (Some(9), 2, 9),
            (Some(2), 9, 9),
            (Some(u64::MAX), 0, u64::MAX),
        ] {
            let allocation = IdentifierRequest::Observe(observed)
                .prepare(current)
                .unwrap();
            assert_eq!(allocation.watermark(), expected);
            assert_eq!(allocation.range(), None);
        }
    }
}