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    /// 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.
32    fn observe_identifier(
33        &self,
34        namespace: &[u8],
35        value: u64,
36    ) -> StorageBackendResult<ObservedIdentifier> {
37        self.allocate_identifiers(namespace, IdentifierRequest::Observe(value))
38            .map(|allocation| ObservedIdentifier::Observed {
39                previous: allocation.previous(),
40            })
41    }
42}
43
44/// What a session learned when it raised a watermark to an identity.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum ObservedIdentifier {
47    /// A watermark the session had read already covers the identity. Where the watermark stood before is not known.
48    Covered,
49    /// The watermark was read and raised under the provider's write admission. `previous` is where it stood, `None` for a namespace that had none.
50    Observed { previous: Option<u64> },
51}
52
53impl ObservedIdentifier {
54    /// 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.
55    pub fn unused_before(self, identity: u64) -> bool {
56        match self {
57            Self::Covered => false,
58            Self::Observed { previous } => previous.is_none_or(|previous| identity > previous),
59        }
60    }
61}
62
63/// 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.
64#[derive(Debug, Clone, Copy)]
65pub enum IdentifierRequest {
66    Observe(u64),
67    Reserve {
68        minimum: u64,
69        maximum: u64,
70        count: NonZeroU64,
71    },
72}
73
74/// The new durable high watermark and, for a reservation, its inclusive allocated range.
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub struct IdentifierAllocation {
77    watermark: u64,
78    first: Option<u64>,
79    previous: Option<u64>,
80}
81
82impl IdentifierAllocation {
83    pub const fn watermark(self) -> u64 {
84        self.watermark
85    }
86
87    /// The watermark before this allocation, or `None` when the namespace had none.
88    pub const fn previous(self) -> Option<u64> {
89        self.previous
90    }
91
92    pub fn range(self) -> Option<RangeInclusive<u64>> {
93        self.first.map(|first| first..=self.watermark)
94    }
95}
96
97impl IdentifierRequest {
98    /// Compute the next state while the provider holds exclusive physical admission. Neither observation nor reservation ever lowers an existing watermark; exhaustion leaves it unchanged.
99    pub fn prepare(self, current: Option<u64>) -> VersionResult<IdentifierAllocation> {
100        let (watermark, first) = match self {
101            Self::Observe(value) => (current.map_or(value, |current| current.max(value)), None),
102            Self::Reserve {
103                minimum,
104                maximum,
105                count,
106            } => {
107                if minimum > maximum {
108                    return Err(VersionError::InvalidIdentifierBounds);
109                }
110                let first = match current {
111                    Some(current) => current
112                        .checked_add(1)
113                        .ok_or(VersionError::IdentifiersExhausted)?
114                        .max(minimum),
115                    None => minimum,
116                };
117                let last = first
118                    .checked_add(count.get() - 1)
119                    .filter(|last| *last <= maximum)
120                    .ok_or(VersionError::IdentifiersExhausted)?;
121                (last, Some(first))
122            }
123        };
124        Ok(IdentifierAllocation {
125            watermark,
126            first,
127            previous: current,
128        })
129    }
130
131    /// Validate a request before physical admission and charge the provider's key binding/copy workspace. No identifier is consumed if this fails.
132    pub fn reserve_workspace(
133        self,
134        namespace: &[u8],
135        control: &StorageReadControl,
136    ) -> VersionResult<MemoryReservation> {
137        self.prepare(None)?;
138        reserve_identifier_workspace(namespace, control)
139    }
140}
141
142/// Charge namespace bindings for either a reservation or a read without retaining the caller's bytes.
143pub fn reserve_identifier_workspace(
144    namespace: &[u8],
145    control: &StorageReadControl,
146) -> VersionResult<MemoryReservation> {
147    control.cancellation().check()?;
148    if namespace.is_empty() {
149        return Err(VersionError::InvalidEncoding("empty identifier namespace"));
150    }
151    if namespace == RECLAMATION_EPOCH_NAMESPACE || namespace.starts_with(RECLAMATION_DOMAIN_PREFIX)
152    {
153        return Err(VersionError::InvalidEncoding(
154            "reserved reclamation namespace",
155        ));
156    }
157    let bytes = namespace
158        .len()
159        .checked_mul(2)
160        .and_then(|bytes| bytes.checked_add(64))
161        .ok_or(MemoryError::SizeOverflow)?;
162    Ok(control.memory().reserve(bytes)?)
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168
169    fn reserve(minimum: u64, maximum: u64, count: u64) -> IdentifierRequest {
170        IdentifierRequest::Reserve {
171            minimum,
172            maximum,
173            count: NonZeroU64::new(count).unwrap(),
174        }
175    }
176
177    #[test]
178    fn reservations_cover_zero_full_width_bounds_and_atomic_exhaustion() {
179        assert_eq!(reserve(0, 9, 2).prepare(None).unwrap().range(), Some(0..=1));
180        assert_eq!(
181            reserve(5, 9, 2).prepare(Some(1)).unwrap().range(),
182            Some(5..=6)
183        );
184        assert_eq!(
185            reserve(0, u64::MAX, 2)
186                .prepare(Some(u64::MAX - 2))
187                .unwrap()
188                .range(),
189            Some(u64::MAX - 1..=u64::MAX)
190        );
191        for (request, current) in [
192            (reserve(0, 9, 2), Some(8)),
193            (reserve(0, u64::MAX, 2), Some(u64::MAX - 1)),
194            (reserve(0, u64::MAX, 1), Some(u64::MAX)),
195        ] {
196            assert!(matches!(
197                request.prepare(current),
198                Err(VersionError::IdentifiersExhausted)
199            ));
200        }
201        assert!(matches!(
202            reserve(2, 1, 1).prepare(None),
203            Err(VersionError::InvalidIdentifierBounds)
204        ));
205    }
206
207    #[test]
208    fn observations_never_rewind_or_wrap_a_namespace() {
209        for (current, observed, expected) in [
210            (None, 0, 0),
211            (Some(9), 2, 9),
212            (Some(2), 9, 9),
213            (Some(u64::MAX), 0, u64::MAX),
214        ] {
215            let allocation = IdentifierRequest::Observe(observed)
216                .prepare(current)
217                .unwrap();
218            assert_eq!(allocation.watermark(), expected);
219            assert_eq!(allocation.range(), None);
220        }
221    }
222}
uqa_storage/mvcc/identifiers.rs

uqa_storage/mvcc/
identifiers.rs
