uqa_storage/mvcc/
commit.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Private evaluated changes and provider-independent revision validation.
8
9mod requirements;
10pub(super) use requirements::RecordRequirement;
11
12use std::collections::BTreeMap;
13use std::sync::Arc;
14
15use sha2::{Digest, Sha256};
16use uqa_core::memory::{BudgetedVec, MemoryError};
17use uqa_core::CancellationToken;
18
19use crate::read_control::StorageReadControl;
20
21use super::graph::{GraphEffects, OwnedGraphMutation};
22use super::key::RecordKey;
23use super::{
24    CommitFingerprint, CommitSequence, GraphMutation, RecordWrite, VersionError, VersionResult,
25};
26
27#[derive(Clone)]
28pub struct PreparedRecordWrite {
29    key: RecordKey,
30    expected: Option<CommitSequence>,
31    value: Option<Arc<BudgetedVec<u8>>>,
32    kind: RecordWriteKind,
33}
34
35#[derive(Clone, Copy, PartialEq, Eq)]
36pub(crate) enum RecordWriteKind {
37    Canonical,
38    GraphCache,
39    GraphPreview,
40    Occurrence,
41    OccurrenceCache,
42    IVFPreview,
43    HNSWPreview,
44    Marker,
45    StatisticsMaintenance,
46}
47
48impl PreparedRecordWrite {
49    pub(super) fn from_shared(
50        key: RecordKey,
51        expected: Option<CommitSequence>,
52        value: Option<Arc<BudgetedVec<u8>>>,
53    ) -> Self {
54        Self {
55            key,
56            expected,
57            value,
58            kind: RecordWriteKind::Canonical,
59        }
60    }
61
62    pub(super) fn copy_bytes(
63        key: &[u8],
64        expected: Option<CommitSequence>,
65        value: Option<&[u8]>,
66        control: &StorageReadControl,
67    ) -> VersionResult<Self> {
68        control.cancellation().check()?;
69        let key = RecordKey::new(key, control.memory())?;
70        let value = value
71            .map(|value| {
72                let mut owned = BudgetedVec::new(control.memory());
73                owned.extend_from_slice(value)?;
74                Ok::<_, VersionError>(Arc::new(owned))
75            })
76            .transpose()?;
77        Ok(Self::from_shared(key, expected, value))
78    }
79
80    pub fn key(&self) -> &[u8] {
81        self.key.bytes()
82    }
83
84    pub fn expected(&self) -> Option<CommitSequence> {
85        self.expected
86    }
87
88    pub fn value(&self) -> Option<&[u8]> {
89        self.value.as_ref().map(|value| &***value)
90    }
91
92    pub(crate) fn shared_value(&self) -> Option<Arc<BudgetedVec<u8>>> {
93        self.value.clone()
94    }
95
96    pub(super) fn shared_key(&self) -> RecordKey {
97        self.key.clone()
98    }
99
100    pub(super) fn rebase(mut self, expected: Option<CommitSequence>) -> Self {
101        self.expected = expected;
102        self
103    }
104
105    pub(crate) fn kind(&self) -> RecordWriteKind {
106        self.kind
107    }
108    pub(crate) fn with_kind(mut self, kind: RecordWriteKind) -> Self {
109        self.kind = kind;
110        self
111    }
112
113    pub(super) fn with_value(mut self, value: BudgetedVec<u8>) -> Self {
114        self.value = Some(Arc::new(value));
115        self
116    }
117}
118
119/// Immutable prepared replacements. Construct before opening a physical writer.
120pub struct PreparedRecordCommit {
121    writes: BudgetedVec<PreparedRecordWrite>,
122    fingerprint: CommitFingerprint,
123    pub(super) graph: Option<GraphEffects>,
124    pub(super) vector: Option<super::vector::VectorEffects>,
125    pub(super) notification: Option<Arc<super::notifications::NotificationEffect>>,
126    pub(super) resolved_at: Option<CommitSequence>,
127    requirements: Option<Arc<BudgetedVec<RecordRequirement>>>,
128}
129
130impl PreparedRecordCommit {
131    pub fn new(writes: &[RecordWrite<'_>], control: &StorageReadControl) -> VersionResult<Self> {
132        control.cancellation().check()?;
133        let slots = writes
134            .len()
135            .checked_mul(std::mem::size_of::<(&[u8], usize)>())
136            .ok_or(MemoryError::SizeOverflow)?;
137        // Charge logical tree entries; allocator node bookkeeping is separate.
138        let seen_memory = control.memory().reserve(slots)?;
139        let mut seen = BTreeMap::new();
140        for (index, write) in writes.iter().enumerate() {
141            control.cancellation().check()?;
142            if let Some(first) = seen.insert(write.key, index) {
143                return Err(VersionError::DuplicateRecord {
144                    first,
145                    second: index,
146                });
147            }
148        }
149        drop(seen);
150        drop(seen_memory);
151
152        let mut prepared = BudgetedVec::new(control.memory());
153        prepared.reserve(writes.len())?;
154        for write in writes {
155            prepared.push(PreparedRecordWrite::copy_bytes(
156                write.key,
157                write.expected,
158                write.value,
159                control,
160            )?)?;
161        }
162        control.cancellation().check()?;
163        Self::from_unique_owned(prepared, control)
164    }
165
166    pub fn records(&self) -> &[PreparedRecordWrite] {
167        &self.writes
168    }
169
170    /// The caller supplies exactly one final replacement for each identity.
171    pub(super) fn from_unique_owned(
172        writes: BudgetedVec<PreparedRecordWrite>,
173        control: &StorageReadControl,
174    ) -> VersionResult<Self> {
175        let mut digest = Sha256::new();
176        let typed = writes
177            .iter()
178            .any(|write| write.kind != RecordWriteKind::Canonical);
179        if typed {
180            digest.update(b"UQA prepared record kinds 1");
181        } else {
182            digest.update(b"UQA prepared records 1");
183        }
184        digest.update((writes.len() as u64).to_be_bytes());
185        for write in writes.iter() {
186            control.cancellation().check()?;
187            if typed {
188                digest.update([match write.kind {
189                    RecordWriteKind::Canonical => 0,
190                    RecordWriteKind::GraphCache => 1,
191                    RecordWriteKind::GraphPreview => 2,
192                    RecordWriteKind::Occurrence => 3,
193                    RecordWriteKind::OccurrenceCache => 4,
194                    RecordWriteKind::IVFPreview => 5,
195                    RecordWriteKind::HNSWPreview => 6,
196                    RecordWriteKind::Marker => 7,
197                    RecordWriteKind::StatisticsMaintenance => 8,
198                }]);
199            }
200            digest.update((write.key().len() as u64).to_be_bytes());
201            hash_bytes(&mut digest, write.key(), control)?;
202            digest.update(
203                write
204                    .expected()
205                    .map_or(0, CommitSequence::as_u64)
206                    .to_be_bytes(),
207            );
208            digest.update([
209                u8::from(write.expected().is_some()),
210                u8::from(write.value().is_some()),
211            ]);
212            if let Some(value) = write.value() {
213                digest.update((value.len() as u64).to_be_bytes());
214                hash_bytes(&mut digest, value, control)?;
215            }
216        }
217        control.cancellation().check()?;
218        Ok(Self {
219            writes,
220            fingerprint: digest.finalize().into(),
221            graph: None,
222            vector: None,
223            notification: None,
224            resolved_at: None,
225            requirements: None,
226        })
227    }
228
229    pub(super) fn with_graph_effects(
230        mut self,
231        base: CommitSequence,
232        operations: &[OwnedGraphMutation],
233        control: &StorageReadControl,
234    ) -> VersionResult<Self> {
235        if operations.is_empty()
236            && self.writes.iter().all(|write| {
237                !matches!(
238                    write.kind,
239                    RecordWriteKind::GraphCache | RecordWriteKind::GraphPreview
240                )
241            })
242        {
243            return Ok(self);
244        }
245        let mut owned = BudgetedVec::new(control.memory());
246        owned.reserve(operations.len())?;
247        let mut digest = Sha256::new();
248        digest.update(b"UQA prepared graph effects 1");
249        digest.update(self.fingerprint);
250        digest.update(base.as_u64().to_be_bytes());
251        digest.update((operations.len() as u64).to_be_bytes());
252        let mut text = |bytes: &[u8]| -> VersionResult<()> {
253            digest.update((bytes.len() as u64).to_be_bytes());
254            hash_bytes(&mut digest, bytes, control)
255        };
256        for operation in operations {
257            control.cancellation().check()?;
258            match operation.borrowed() {
259                GraphMutation::InvalidateGraph(graph) => {
260                    text(b"graph")?;
261                    text(graph.as_bytes())?;
262                }
263                GraphMutation::InvalidatePath(index) => {
264                    text(b"invalidate-path")?;
265                    text(index.as_bytes())?;
266                }
267                GraphMutation::InvalidateEntity(kind, id) => {
268                    text(b"entity")?;
269                    text(kind.as_str().as_bytes())?;
270                    text(&id.to_be_bytes())?;
271                }
272                GraphMutation::PublishPath {
273                    index,
274                    graph,
275                    definition,
276                } => {
277                    text(b"path")?;
278                    text(index.as_bytes())?;
279                    text(graph.as_bytes())?;
280                    text(definition.as_bytes())?;
281                }
282            }
283            owned.push(operation.clone())?;
284        }
285        self.fingerprint = digest.finalize().into();
286        self.graph = Some(GraphEffects {
287            base,
288            operations: owned,
289        });
290        Ok(self)
291    }
292
293    pub(super) fn with_notification_effect(
294        mut self,
295        effect: Option<&Arc<super::notifications::NotificationEffect>>,
296    ) -> Self {
297        if let Some(effect) = effect {
298            self.fingerprint = effect.seal(self.fingerprint);
299            self.notification = Some(Arc::clone(effect));
300        }
301        self
302    }
303
304    pub(crate) fn resolved(mut self, original: &Self, sequence: CommitSequence) -> Self {
305        self.fingerprint = original.fingerprint;
306        self.resolved_at = Some(sequence);
307        self.requirements.clone_from(&original.requirements);
308        self
309    }
310
311    pub(super) fn seal_vector_effects(
312        &mut self,
313        fingerprint: CommitFingerprint,
314        effects: super::vector::VectorEffects,
315    ) {
316        self.fingerprint = fingerprint;
317        self.vector = Some(effects);
318    }
319
320    pub(crate) fn retain_graph_effects(
321        self,
322        original: &Self,
323        control: &StorageReadControl,
324    ) -> VersionResult<Self> {
325        match &original.graph {
326            Some(effects) => self.with_graph_effects(effects.base, &effects.operations, control),
327            None => Ok(self),
328        }
329    }
330
331    /// Check the snapshot used to discover derived dependencies under exclusive admission, after resolving any existing receipt and before validating record heads. A mismatch proves this pending attempt has not published and permits pure effect preparation to restart.
332    pub fn validate_snapshot(&self, current: CommitSequence) -> VersionResult<()> {
333        if self.graph.is_some()
334            || self.vector.is_some()
335            || self.notification.is_some()
336            || self
337                .writes
338                .iter()
339                .any(|write| write.kind != RecordWriteKind::Canonical)
340        {
341            return Err(VersionError::InvalidEncoding(
342                "unresolved storage commit effects",
343            ));
344        }
345        if let Some(expected) = self.resolved_at {
346            if expected != current {
347                return Err(VersionError::CommitSnapshotChanged {
348                    expected,
349                    actual: current,
350                });
351            }
352        }
353        Ok(())
354    }
355
356    pub fn fingerprint(&self) -> CommitFingerprint {
357        self.fingerprint
358    }
359
360    /// Check all preconditions under the provider's exclusive commit boundary. The callback reads current committed heads, not the caller's old snapshot.
361    pub fn validate(
362        &self,
363        cancellation: &CancellationToken,
364        mut head_revision: impl FnMut(&[u8]) -> VersionResult<Option<CommitSequence>>,
365    ) -> VersionResult<()> {
366        cancellation.check()?;
367        if self.graph.is_some()
368            || self.vector.is_some()
369            || self.notification.is_some()
370            || self
371                .writes
372                .iter()
373                .any(|write| write.kind != RecordWriteKind::Canonical)
374        {
375            return Err(VersionError::InvalidEncoding(
376                "unresolved storage commit effects",
377            ));
378        }
379        for (index, write) in self.writes.iter().enumerate() {
380            cancellation.check()?;
381            let actual = head_revision(write.key())?;
382            if write.expected != actual {
383                return Err(VersionError::WriteConflict {
384                    mutation: index,
385                    expected: write.expected,
386                    actual,
387                });
388            }
389        }
390        self.validate_requirements(cancellation, head_revision)?;
391        cancellation.check()?;
392        Ok(())
393    }
394}
395
396fn hash_bytes(
397    digest: &mut Sha256,
398    bytes: &[u8],
399    control: &StorageReadControl,
400) -> VersionResult<()> {
401    for chunk in bytes.chunks(65536) {
402        control.cancellation().check()?;
403        digest.update(chunk);
404    }
405    Ok(())
406}
uqa_storage/mvcc/commit.rs

uqa_storage/mvcc/
commit.rs
