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

uqa_storage/mvcc/
commit.rs
