uqa_storage/mvcc/
history.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Ordered immutable revisions with one retained predecessor at the horizon.
8
9use std::sync::Arc;
10use uqa_core::memory::{BudgetedVec, MemoryBudget};
11
12use super::{CommitSequence, VersionError, VersionResult};
13
14pub type SharedRecordValue = Arc<BudgetedVec<u8>>;
15
16pub struct ScannedRecord {
17    pub key: BudgetedVec<u8>,
18    pub version: RecordVersion<SharedRecordValue>,
19}
20
21#[derive(Debug, Clone)]
22pub struct RecordVersion<T> {
23    sequence: CommitSequence,
24    value: Option<T>,
25}
26
27impl<T> RecordVersion<T> {
28    pub fn new(sequence: CommitSequence, value: Option<T>) -> VersionResult<Self> {
29        if sequence == CommitSequence::INITIAL {
30            return Err(VersionError::InvalidEncoding(
31                "record version has the initial sequence",
32            ));
33        }
34        Ok(Self { sequence, value })
35    }
36    pub fn sequence(&self) -> CommitSequence {
37        self.sequence
38    }
39
40    /// A missing payload denotes deletion; the revision itself remains present.
41    pub fn value(&self) -> Option<&T> {
42        self.value.as_ref()
43    }
44
45    pub fn into_parts(self) -> (CommitSequence, Option<T>) {
46        (self.sequence, self.value)
47    }
48}
49
50impl RecordVersion<SharedRecordValue> {
51    /// Own provider bytes under the read allowance before closing a physical read window.
52    pub fn copy_bytes(
53        sequence: CommitSequence,
54        value: Option<&[u8]>,
55        control: &crate::read_control::StorageReadControl,
56    ) -> VersionResult<Self> {
57        control.cancellation().check()?;
58        let mut record = Self::new(sequence, None)?;
59        record.value = value
60            .map(|bytes| {
61                let mut value = BudgetedVec::new(control.memory());
62                value.extend_from_slice(bytes)?;
63                Ok::<_, VersionError>(Arc::new(value))
64            })
65            .transpose()?;
66        Ok(record)
67    }
68}
69
70impl ScannedRecord {
71    pub fn copy_key(
72        key: &[u8],
73        version: RecordVersion<SharedRecordValue>,
74        control: &crate::read_control::StorageReadControl,
75    ) -> VersionResult<Self> {
76        control.cancellation().check()?;
77        let mut owned = BudgetedVec::new(control.memory());
78        owned.extend_from_slice(key)?;
79        Ok(Self {
80            key: owned,
81            version,
82        })
83    }
84}
85
86/// A record's committed revisions. Payload allocations retain separate owners.
87#[derive(Debug)]
88pub struct RecordHistory<T> {
89    versions: BudgetedVec<RecordVersion<T>>,
90}
91
92impl<T> RecordHistory<T> {
93    pub fn new(memory: &MemoryBudget) -> Self {
94        Self {
95            versions: BudgetedVec::new(memory),
96        }
97    }
98
99    pub fn len(&self) -> usize {
100        self.versions.len()
101    }
102
103    pub fn is_empty(&self) -> bool {
104        self.versions.is_empty()
105    }
106
107    pub fn head(&self) -> Option<&RecordVersion<T>> {
108        self.versions.last()
109    }
110
111    pub fn visible_at(&self, snapshot: CommitSequence) -> Option<&RecordVersion<T>> {
112        let position = self
113            .versions
114            .partition_point(|version| version.sequence <= snapshot);
115        position
116            .checked_sub(1)
117            .map(|position| &self.versions[position])
118    }
119
120    pub fn append(&mut self, sequence: CommitSequence, value: Option<T>) -> VersionResult<()> {
121        self.check_sequence(sequence)?;
122        self.versions.push(RecordVersion { sequence, value })?;
123        Ok(())
124    }
125
126    fn check_sequence(&self, sequence: CommitSequence) -> VersionResult<()> {
127        let previous = self
128            .head()
129            .map_or(CommitSequence::INITIAL, RecordVersion::sequence);
130        if sequence <= previous {
131            return Err(VersionError::CommitOrder {
132                previous,
133                next: sequence,
134            });
135        }
136        Ok(())
137    }
138
139    /// Retain the newest revision at/before the horizon and every later one.
140    ///
141    /// The caller must hold snapshot admission while choosing and applying the horizon. Tombstones are retained as revisions, including at the head.
142    ///
143    /// Eligible revisions are already removed if shrinking fails; the retained versions stay valid and a later call can retry releasing excess capacity.
144    pub fn reclaim_before(&mut self, horizon: CommitSequence) -> VersionResult<usize> {
145        let removed = self
146            .versions
147            .partition_point(|version| version.sequence <= horizon)
148            .saturating_sub(1);
149        if removed != 0 {
150            let retained = self.versions.len() - removed;
151            self.versions.rotate_left(removed);
152            self.versions.truncate(retained);
153        }
154        self.versions.shrink_to_fit()?;
155        Ok(removed)
156    }
157}
158
159impl<T: Clone> RecordHistory<T> {
160    /// Prepare a replacement history without mutating the published owner.
161    pub fn fork_appending(
162        &self,
163        sequence: CommitSequence,
164        value: Option<T>,
165    ) -> VersionResult<Self> {
166        self.check_sequence(sequence)?;
167        let mut candidate = Self::new(self.versions.budget());
168        let capacity = self
169            .versions
170            .len()
171            .checked_add(1)
172            .ok_or(uqa_core::memory::MemoryError::SizeOverflow)?;
173        candidate.versions.reserve(capacity)?;
174        for version in self.versions.iter() {
175            candidate.versions.push(version.clone())?;
176        }
177        candidate.append(sequence, value)?;
178        Ok(candidate)
179    }
180}
uqa_storage/mvcc/history.rs

uqa_storage/mvcc/
history.rs
