uqa_storage/mvcc/
identifiers.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Monotonic identifier reservations are independent of record visibility and transaction undo.
8
9mod conformance;
10pub use conformance::{verify_identifier_allocations, verify_identifier_batches};
11
12use std::{num::NonZeroU64, ops::RangeInclusive};
13
14use uqa_core::memory::{MemoryError, MemoryReservation};
15
16use crate::{read_control::StorageReadControl, StorageBackendResult};
17
18use super::{VersionError, VersionResult, RECLAMATION_DOMAIN_PREFIX, RECLAMATION_EPOCH_NAMESPACE};
19
20/// 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.
21pub trait IdentifierAllocator: Send + Sync {
22    /// 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.
23    fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>>;
24
25    fn allocate_identifiers(
26        &self,
27        namespace: &[u8],
28        request: IdentifierRequest,
29    ) -> StorageBackendResult<IdentifierAllocation>;
30}
31
32/// 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.
33#[derive(Debug, Clone, Copy)]
34pub enum IdentifierRequest {
35    Observe(u64),
36    Reserve {
37        minimum: u64,
38        maximum: u64,
39        count: NonZeroU64,
40    },
41}
42
43/// The new durable high watermark and, for a reservation, its inclusive allocated range.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub struct IdentifierAllocation {
46    watermark: u64,
47    first: Option<u64>,
48}
49
50impl IdentifierAllocation {
51    pub const fn watermark(self) -> u64 {
52        self.watermark
53    }
54
55    pub fn range(self) -> Option<RangeInclusive<u64>> {
56        self.first.map(|first| first..=self.watermark)
57    }
58}
59
60impl IdentifierRequest {
61    /// Compute the next state while the provider holds exclusive physical admission. Neither observation nor reservation ever lowers an existing watermark; exhaustion leaves it unchanged.
62    pub fn prepare(self, current: Option<u64>) -> VersionResult<IdentifierAllocation> {
63        let (watermark, first) = match self {
64            Self::Observe(value) => (current.map_or(value, |current| current.max(value)), None),
65            Self::Reserve {
66                minimum,
67                maximum,
68                count,
69            } => {
70                if minimum > maximum {
71                    return Err(VersionError::InvalidIdentifierBounds);
72                }
73                let first = match current {
74                    Some(current) => current
75                        .checked_add(1)
76                        .ok_or(VersionError::IdentifiersExhausted)?
77                        .max(minimum),
78                    None => minimum,
79                };
80                let last = first
81                    .checked_add(count.get() - 1)
82                    .filter(|last| *last <= maximum)
83                    .ok_or(VersionError::IdentifiersExhausted)?;
84                (last, Some(first))
85            }
86        };
87        Ok(IdentifierAllocation { watermark, first })
88    }
89
90    /// Validate a request before physical admission and charge the provider's key binding/copy workspace. No identifier is consumed if this fails.
91    pub fn reserve_workspace(
92        self,
93        namespace: &[u8],
94        control: &StorageReadControl,
95    ) -> VersionResult<MemoryReservation> {
96        self.prepare(None)?;
97        reserve_identifier_workspace(namespace, control)
98    }
99}
100
101/// Charge namespace bindings for either a reservation or a read without retaining the caller's bytes.
102pub fn reserve_identifier_workspace(
103    namespace: &[u8],
104    control: &StorageReadControl,
105) -> VersionResult<MemoryReservation> {
106    control.cancellation().check()?;
107    if namespace.is_empty() {
108        return Err(VersionError::InvalidEncoding("empty identifier namespace"));
109    }
110    if namespace == RECLAMATION_EPOCH_NAMESPACE || namespace.starts_with(RECLAMATION_DOMAIN_PREFIX)
111    {
112        return Err(VersionError::InvalidEncoding(
113            "reserved reclamation namespace",
114        ));
115    }
116    let bytes = namespace
117        .len()
118        .checked_mul(2)
119        .and_then(|bytes| bytes.checked_add(64))
120        .ok_or(MemoryError::SizeOverflow)?;
121    Ok(control.memory().reserve(bytes)?)
122}
123
124#[cfg(test)]
125mod tests {
126    use super::*;
127
128    fn reserve(minimum: u64, maximum: u64, count: u64) -> IdentifierRequest {
129        IdentifierRequest::Reserve {
130            minimum,
131            maximum,
132            count: NonZeroU64::new(count).unwrap(),
133        }
134    }
135
136    #[test]
137    fn reservations_cover_zero_full_width_bounds_and_atomic_exhaustion() {
138        assert_eq!(reserve(0, 9, 2).prepare(None).unwrap().range(), Some(0..=1));
139        assert_eq!(
140            reserve(5, 9, 2).prepare(Some(1)).unwrap().range(),
141            Some(5..=6)
142        );
143        assert_eq!(
144            reserve(0, u64::MAX, 2)
145                .prepare(Some(u64::MAX - 2))
146                .unwrap()
147                .range(),
148            Some(u64::MAX - 1..=u64::MAX)
149        );
150        for (request, current) in [
151            (reserve(0, 9, 2), Some(8)),
152            (reserve(0, u64::MAX, 2), Some(u64::MAX - 1)),
153            (reserve(0, u64::MAX, 1), Some(u64::MAX)),
154        ] {
155            assert!(matches!(
156                request.prepare(current),
157                Err(VersionError::IdentifiersExhausted)
158            ));
159        }
160        assert!(matches!(
161            reserve(2, 1, 1).prepare(None),
162            Err(VersionError::InvalidIdentifierBounds)
163        ));
164    }
165
166    #[test]
167    fn observations_never_rewind_or_wrap_a_namespace() {
168        for (current, observed, expected) in [
169            (None, 0, 0),
170            (Some(9), 2, 9),
171            (Some(2), 9, 9),
172            (Some(u64::MAX), 0, u64::MAX),
173        ] {
174            let allocation = IdentifierRequest::Observe(observed)
175                .prepare(current)
176                .unwrap();
177            assert_eq!(allocation.watermark(), expected);
178            assert_eq!(allocation.range(), None);
179        }
180    }
181}
uqa_storage/mvcc/identifiers.rs

uqa_storage/mvcc/
identifiers.rs
